From d82ab0a0ade3c7f672f4014982ae575c503eac01 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 24 Oct 2026 07:37:25 +0200 Subject: [PATCH] perf(paths): batch canonical resolved-path lookups --- cmd/server/paths_batch_2224_test.go | 466 ++++++++++++++++++++++++++++ cmd/server/resolved_index.go | 174 ++++++++++- cmd/server/routes.go | 7 +- 3 files changed, 637 insertions(+), 10 deletions(-) create mode 100644 cmd/server/paths_batch_2224_test.go diff --git a/cmd/server/paths_batch_2224_test.go b/cmd/server/paths_batch_2224_test.go new file mode 100644 index 000000000..057f709ec --- /dev/null +++ b/cmd/server/paths_batch_2224_test.go @@ -0,0 +1,466 @@ +package main + +import ( + "context" + "database/sql" + "database/sql/driver" + "encoding/json" + "errors" + "fmt" + "net/http/httptest" + "path/filepath" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" +) + +func batch2224Store(t testing.TB, counting bool) *PacketStore { + t.Helper() + driverName := "sqlite" + if counting { + registerCountingDriver() + clearBulkTestFailAfterQueries() + driverName = "sqlite-counting-ping-score-bulk" + } + conn, err := sql.Open(driverName, ":memory:") + if err != nil { + t.Fatal(err) + } + conn.SetMaxOpenConns(1) + t.Cleanup(func() { conn.Close(); clearBulkTestFailAfterQueries() }) + if _, err := conn.Exec(`CREATE TABLE observations (id INTEGER PRIMARY KEY, transmission_id INTEGER, path_json TEXT, resolved_path TEXT); CREATE INDEX observations_tx ON observations(transmission_id)`); err != nil { + t.Fatal(err) + } + s := &PacketStore{db: &DB{conn: conn}} + s.initResolvedPathIndex() + return s +} + +func seed2224(t testing.TB, s *PacketStore, n int, fallback bool) map[int][]*StoreObs { + t.Helper() + stmt, err := s.db.conn.Prepare(`INSERT INTO observations VALUES (?,?,?,?)`) + if err != nil { + t.Fatal(err) + } + defer stmt.Close() + out := make(map[int][]*StoreObs, n) + for id := 1; id <= n; id++ { + first := 2*id - 1 + var primary interface{} = `["primary"]` + if fallback { + primary = nil + } + if _, err := stmt.Exec(first, id, `["aa","bb"]`, primary); err != nil { + t.Fatal(err) + } + if _, err := stmt.Exec(first+1, id, `["cc"]`, `["fallback"]`); err != nil { + t.Fatal(err) + } + out[id] = []*StoreObs{{ID: first, PathJSON: `["aa","bb"]`}, {ID: first + 1, PathJSON: `["cc"]`}} + } + return out +} + +func TestCanonicalPaths2224BatchCount(t *testing.T) { + for _, fallback := range []bool{false, true} { + t.Run(fmt.Sprint(fallback), func(t *testing.T) { + s := batch2224Store(t, true) + const n = 1001 // exceeds both 500 and SQLite's older 999-variable budget + snapshots := seed2224(t, s, n, fallback) + resetBulkTestQueryLog() + for tx, obs := range snapshots { + s.fetchResolvedPathForTxBestAt(tx, obs, s.lruNow()) + } + wantScalar := n + if fallback { + wantScalar *= 2 + } + if count := len(bulkTestQueryLog()); count != wantScalar { + t.Fatalf("scalar instrumentation captured %d queries, want %d", count, wantScalar) + } + resetLRU(s) + resetBulkTestQueryLog() + got := s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + wantQueries := 3 + if fallback { + wantQueries = 6 + } + if queries := bulkTestQueryLog(); len(queries) != wantQueries { + t.Fatalf("%d candidates: %d SQL queries, want %d batches", n, len(queries), wantQueries) + } else { + for _, q := range queries { + if q.argCount > 500 { + t.Fatalf("unbounded SQL arguments: %d", q.argCount) + } + } + } + if len(got) != n { + t.Fatalf("got %d paths, want %d", len(got), n) + } + want := "primary" + if fallback { + want = "fallback" + } + for tx, rp := range got { + if len(rp) != 1 || rp[0] == nil || *rp[0] != want { + t.Fatalf("tx %d: %+v, want %s", tx, rp, want) + } + } + }) + } +} + +func TestNodePaths2224UsesBatches(t *testing.T) { + srv, router := setupTestServer(t) + const n = 1001 + for i := 0; i < n; i++ { + txID := seedConfirmTx(t, srv, fmt.Sprintf("batch2224_%d", i), `["aa"]`, `["`+confirmTestTarget+`"]`) + if _, err := srv.db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?,1,'["aa","bb"]',?)`, txID, time.Now().Unix()); err != nil { + t.Fatal(err) + } + } + // Use the existing real-driver query recorder against a private copy of + // this full route fixture. No production connection or live host is used. + file := filepath.Join(t.TempDir(), "paths.db") + if _, err := srv.db.conn.Exec(`VACUUM INTO ?`, file); err != nil { + t.Fatal(err) + } + registerCountingDriver() + conn, err := sql.Open("sqlite-counting-ping-score-bulk", file) + if err != nil { + t.Fatal(err) + } + conn.SetMaxOpenConns(1) + old := srv.db.conn + srv.db.conn = conn + t.Cleanup(func() { conn.Close(); old.Close() }) + reloadConfirmStore(t, srv) + resetBulkTestQueryLog() + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/api/nodes/"+confirmTestTarget+"/paths", nil)) + if w.Code != 200 { + t.Fatalf("status %d: %s", w.Code, w.Body.String()) + } + var response confirmPathsResp + if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil || response.TotalTransmissions < n { + t.Fatalf("response lost candidates: total %d, err %v", response.TotalTransmissions, err) + } + batches := 0 + for _, q := range bulkTestQueryLog() { + if strings.Contains(q.sql, "SELECT resolved_path FROM observations WHERE id =") || strings.Contains(q.sql, "SELECT id, resolved_path FROM observations WHERE transmission_id =") { + t.Fatalf("handler still uses per-transmission SQL: %s", q.sql) + } + if strings.Contains(q.sql, "SELECT id, resolved_path FROM observations WHERE id IN") || strings.Contains(q.sql, "SELECT id, transmission_id, resolved_path FROM observations") { + batches++ + } + } + if batches < 6 || batches > 8 { + t.Fatalf("handler resolved paths in %d batches, want six plus at most baseline fixture chunks", batches) + } +} + +// A real SQLite driver wrapper injects failure only into the new IN batches. +// Scalar fallback queries must still succeed; it also exercises rows.Err and +// Scan failures after a batch has already produced one valid row. +type fault2224Driver struct { + inner driver.Driver + mode string + faults atomic.Int32 +} +type fault2224Conn struct { + driver.Conn + owner *fault2224Driver +} +type fault2224Rows struct { + driver.Rows + mode string + read bool +} + +var fault2224Names atomic.Int64 + +func (d *fault2224Driver) Open(name string) (driver.Conn, error) { + c, err := d.inner.Open(name) + if err != nil { + return nil, err + } + return &fault2224Conn{Conn: c, owner: d}, nil +} +func (c *fault2224Conn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) { + batch := strings.Contains(query, " IN (") + if batch && (c.owner.mode == "primary" && strings.Contains(query, "WHERE id IN") || c.owner.mode == "fallback" && strings.Contains(query, "transmission_id IN")) { + c.owner.faults.Add(1) + return nil, errors.New("injected batch query failure") + } + rows, err := c.Conn.(driver.QueryerContext).QueryContext(ctx, query, args) + if err == nil && batch && (c.owner.mode == "rows" || c.owner.mode == "scan") { + c.owner.faults.Add(1) + return &fault2224Rows{Rows: rows, mode: c.owner.mode}, nil + } + return rows, err +} +func (r *fault2224Rows) Next(values []driver.Value) error { + if r.read { + if r.mode == "scan" { + values[0] = "not-an-integer" + return nil + } + return errors.New("injected row iteration failure") + } + r.read = true + return r.Rows.Next(values) +} +func TestCanonicalPaths2224BatchFailureFallsBack(t *testing.T) { + for _, mode := range []string{"primary", "fallback", "rows", "scan"} { + t.Run(mode, func(t *testing.T) { + s := batch2224Store(t, false) + inner := s.db.conn.Driver() + // Replace only this test's empty connection, never a shared/global DB. + s.db.conn.Close() + d := &fault2224Driver{inner: inner, mode: mode} + name := fmt.Sprintf("sqlite-paths2224-fault-%d", fault2224Names.Add(1)) + sql.Register(name, d) + conn, err := sql.Open(name, ":memory:") + if err != nil { + t.Fatal(err) + } + conn.SetMaxOpenConns(1) + s.db.conn = conn + t.Cleanup(func() { conn.Close() }) + if _, err := conn.Exec(`CREATE TABLE observations (id INTEGER PRIMARY KEY, transmission_id INTEGER, path_json TEXT, resolved_path TEXT)`); err != nil { + t.Fatal(err) + } + inputs := seed2224(t, s, 3, true) + got := s.loadCanonicalResolvedPathsAt(inputs, s.lruNow()) + if d.faults.Load() == 0 { + t.Fatal("fault was not exercised") + } + if len(got) != 3 { + t.Fatalf("failed batch lost scalar fallback rows: %v", got) + } + for tx, rp := range got { + if len(rp) != 1 || rp[0] == nil || *rp[0] != "fallback" { + t.Fatalf("tx %d: wrong recovered result", tx) + } + } + }) + } +} + +func TestCanonicalPaths2224MatchesBestAt(t *testing.T) { + cases := []struct { + name string + paths []string + stored []interface{} + want []*string + }{ + {"normal", []string{`["a","b"]`, `["a"]`}, []interface{}{`["long"]`, `["short"]`}, []*string{str2224("long")}}, + {"null primary", []string{`["a","b"]`, `["a"]`}, []interface{}{nil, `["short"]`}, []*string{str2224("short")}}, + {"empty string primary", []string{`["a","b"]`, `["a"]`}, []interface{}{``, `["short"]`}, []*string{str2224("short")}}, + {"malformed primary", []string{`["a","b"]`, `["a"]`}, []interface{}{`invalid`, `["short"]`}, []*string{str2224("short")}}, + {"null JSON primary", []string{`["a","b"]`, `["a"]`}, []interface{}{`null`, `["short"]`}, []*string{str2224("short")}}, + {"empty array primary", []string{`["a","b"]`, `["a"]`}, []interface{}{`[]`, `["short"]`}, []*string{}}, + {"nil entries", []string{`["a","b"]`}, []interface{}{`[null,"node"]`}, []*string{nil, str2224("node")}}, + {"snapshot tie order", []string{`["b"]`, `["a"]`}, []interface{}{`["second-id-first"]`, `["first-id-second"]`}, []*string{str2224("second-id-first")}}, + {"fallback tie order", []string{`["a","b"]`, `["b"]`, `["a"]`}, []interface{}{nil, `["second-id-first"]`, `["first-id-second"]`}, []*string{str2224("second-id-first")}}, + {"nothing stored", []string{`["a","b"]`, `["a"]`}, []interface{}{nil, nil}, nil}, + {"malformed raw path", []string{`invalid`, `["a"]`}, []interface{}{`["bad-raw"]`, `["good-raw"]`}, []*string{str2224("good-raw")}}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + s := batch2224Store(t, false) + var obs []*StoreObs + for i, raw := range c.paths { + id := 10 - i // snapshot order deliberately differs from SQLite PK order + if _, err := s.db.conn.Exec(`INSERT INTO observations VALUES (?,?,?,?)`, id, 7, raw, c.stored[i]); err != nil { + t.Fatal(err) + } + obs = append(obs, &StoreObs{ID: id, PathJSON: raw}) + } + // A longer stored path missing from the immutable snapshot must not win. + if _, err := s.db.conn.Exec(`INSERT INTO observations VALUES (99,7,'["x","y","z"]','["not-in-snapshot"]')`); err != nil { + t.Fatal(err) + } + now := s.lruNow() + want := s.fetchResolvedPathForTxBestAt(7, obs, now) + resetLRU(s) + got := s.loadCanonicalResolvedPathsAt(map[int][]*StoreObs{7: obs, 8: nil}, now)[7] + if !reflect.DeepEqual(got, want) || !reflect.DeepEqual(got, c.want) { + t.Fatalf("batch=%v scalar=%v expected=%v", deref2224(got), deref2224(want), deref2224(c.want)) + } + }) + } +} + +func str2224(s string) *string { return &s } +func deref2224(rp []*string) string { + var b strings.Builder + for _, p := range rp { + if p == nil { + b.WriteString(",") + } else { + b.WriteString(*p + ",") + } + } + return b.String() +} + +func TestCanonicalPaths2224LRUAndErrors(t *testing.T) { + s := batch2224Store(t, true) + snapshots := seed2224(t, s, 2, false) + advance := fakeLRUClock(s) + s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + if _, err := s.db.conn.Exec(`UPDATE observations SET resolved_path='["changed"]'`); err != nil { + t.Fatal(err) + } + resetBulkTestQueryLog() + got := s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + if len(bulkTestQueryLog()) != 0 || *got[1][0] != "primary" { + t.Fatal("warm LRU was not used") + } + advance(resolvedPathLRUTTL) + got = s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + if *got[1][0] != "primary" { + t.Fatal("TTL boundary expired at equality instead of strictly after") + } + advance(time.Nanosecond) + got = s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + if *got[1][0] != "changed" { + t.Fatal("expired same-row overwrite remained stale") + } + resetLRU(s) + setBulkTestFailAfterQueries(0) + if got := s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()); len(got) != 0 { + t.Fatal("query errors fabricated canonical paths") + } + clearBulkTestFailAfterQueries() + if got := s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()); len(got) != 2 { + t.Fatal("failed batch poisoned subsequent recovery") + } + if got := (&PacketStore{}).loadCanonicalResolvedPathsAt(snapshots, 0); len(got) != 0 { + t.Fatal("nil DB fabricated paths") + } +} + +func TestCanonicalPaths2224HistoricalRowsNotDecoded(t *testing.T) { + s := batch2224Store(t, false) + inputs := seed2224(t, s, 20, true) + largePath := `[` + strings.TrimSuffix(strings.Repeat(`"outside-snapshot",`, 30), ",") + `]` + for tx := 1; tx <= 20; tx++ { + for extra := 0; extra < 40; extra++ { + if _, err := s.db.conn.Exec(`INSERT INTO observations VALUES (?,?,?,?)`, 1000+40*(tx-1)+extra, tx, `["aa","bb","cc"]`, largePath); err != nil { + t.Fatal(err) + } + } + } + before := testing.AllocsPerRun(1, func() { + resetLRU(s) + out := make(map[int][]*string, len(inputs)) + for id, obs := range inputs { + out[id] = s.fetchResolvedPathForTxBestAt(id, obs, s.lruNow()) + } + batch2224Sink = out + }) + after := testing.AllocsPerRun(1, func() { + resetLRU(s) + batch2224Sink = s.loadCanonicalResolvedPathsAt(inputs, s.lruNow()) + }) + defer func() { batch2224Sink = nil }() + // No wall-clock budget: the regression is retaining/decoding known- + // irrelevant rows. The real driver still allocates their scanned strings. + if after >= before/2 { + t.Fatalf("historical paths still decoded: %.0f allocations, scalar %.0f (want < half)", after, before) + } + if len(batch2224Sink) != len(inputs) { + t.Fatalf("allocation guard lost paths: got %d, want %d", len(batch2224Sink), len(inputs)) + } + for tx, rp := range batch2224Sink { + if len(rp) != 1 || rp[0] == nil || *rp[0] != "fallback" { + t.Fatalf("tx %d: historical row won", tx) + } + } + t.Logf("allocations with 800 out-of-snapshot rows: scalar %.0f, batched %.0f", before, after) +} + +func BenchmarkCanonicalPaths2224(b *testing.B) { + for _, n := range []int{1001, 30000} { + for _, fallback := range []bool{false, true} { + for _, batched := range []bool{false, true} { + b.Run(fmt.Sprintf("tx=%d/fallback=%t/batch=%t", n, fallback, batched), func(b *testing.B) { + s := batch2224Store(b, false) + inputs := seed2224(b, s, n, fallback) + // Five observations per transmission, while the original + // primary remains longest and the first shorter remains best. + for tx := 1; tx <= n; tx++ { + for extra := 0; extra < 3; extra++ { + id := 2*n + 3*(tx-1) + extra + 1 + if _, err := s.db.conn.Exec(`INSERT INTO observations VALUES (?,?,?,?)`, id, tx, `[]`, `["extra"]`); err != nil { + b.Fatal(err) + } + inputs[tx] = append(inputs[tx], &StoreObs{ID: id, PathJSON: `[]`}) + } + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + resetLRU(s) + now := s.lruNow() + if batched { + batch2224Sink = s.loadCanonicalResolvedPathsAt(inputs, now) + } else { + out := make(map[int][]*string, len(inputs)) + for id, obs := range inputs { + if rp := s.fetchResolvedPathForTxBestAt(id, obs, now); rp != nil { + out[id] = rp + } + } + batch2224Sink = out + } + } + b.StopTimer() + batch2224Sink = nil + }) + } + } + } +} + +var batch2224Sink map[int][]*string + +func BenchmarkCanonicalPaths2224HistoricalFanout(b *testing.B) { + for _, batched := range []bool{false, true} { + b.Run(fmt.Sprint(batched), func(b *testing.B) { + s := batch2224Store(b, false) + inputs := seed2224(b, s, 100, true) + largePath := `[` + strings.TrimSuffix(strings.Repeat(`"outside-snapshot",`, 30), ",") + `]` + for tx := 1; tx <= 100; tx++ { + for extra := 0; extra < 200; extra++ { + if _, err := s.db.conn.Exec(`INSERT INTO observations VALUES (?,?,?,?)`, 10000+200*(tx-1)+extra, tx, `["aa","bb","cc"]`, largePath); err != nil { + b.Fatal(err) + } + } + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + resetLRU(s) + now := s.lruNow() + if batched { + batch2224Sink = s.loadCanonicalResolvedPathsAt(inputs, now) + } else { + out := make(map[int][]*string, len(inputs)) + for id, obs := range inputs { + if rp := s.fetchResolvedPathForTxBestAt(id, obs, now); rp != nil { + out[id] = rp + } + } + batch2224Sink = out + } + } + b.StopTimer() + batch2224Sink = nil + }) + } +} diff --git a/cmd/server/resolved_index.go b/cmd/server/resolved_index.go index b603016bb..64535f821 100644 --- a/cmd/server/resolved_index.go +++ b/cmd/server/resolved_index.go @@ -14,6 +14,7 @@ import ( "encoding/json" "hash/fnv" "log" + "sort" "strconv" "strings" "time" @@ -69,20 +70,181 @@ func (s *PacketStore) bestResolvedPathAt(txID, count int, at func(int) resolvedP return rp } paths := s.fetchResolvedPathsForTx(txID) + best, bestID := longestStoredResolvedPath(count, at, paths) + if bestID != 0 { + s.lruMu.Lock() + s.lruPut(bestID, best) + s.lruMu.Unlock() + } + return best +} + +// longestStoredResolvedPath preserves the scalar helper's first-in-snapshot +// tie rule. Rows added after the snapshot never become candidates. +func longestStoredResolvedPath(count int, at func(int) resolvedPathObservation, stored map[int][]*string) ([]*string, int) { var best []*string bestLen, bestID := -1, 0 for i := 0; i < count; i++ { obs := at(i) - if rp := paths[obs.id]; rp != nil && pathLen(obs.pathJSON) > bestLen { + if rp := stored[obs.id]; rp != nil && pathLen(obs.pathJSON) > bestLen { best, bestLen, bestID = rp, pathLen(obs.pathJSON), obs.id } } - if bestID != 0 { - s.lruMu.Lock() - s.lruPut(bestID, best) - s.lruMu.Unlock() + return best, bestID +} + +const resolvedPathBatchSize = 500 // below SQLite's older 999-bound-variable limit + +// loadCanonicalResolvedPathsAt is the batched form of BestAt for /paths. +// The observation slices are immutable append-only snapshots captured under +// s.mu. SQL and LRU access happen only after that lock has been released. +// The supplied clock preserves the existing once-per-request TTL decision. +func (s *PacketStore) loadCanonicalResolvedPathsAt(snapshots map[int][]*StoreObs, now int64) map[int][]*string { + out := make(map[int][]*string, len(snapshots)) + if s.db == nil || s.db.conn == nil { + return out + } + longestTx := make(map[int]int, len(snapshots)) + missing := make([]int, 0, len(snapshots)) + fallback := make([]int, 0) + for txID, observations := range snapshots { + if len(observations) == 0 { + continue + } + longest, longestLen := observations[0], pathLen(observations[0].PathJSON) + for _, obs := range observations[1:] { + if l := pathLen(obs.PathJSON); l > longestLen { + longest, longestLen = obs, l + } + } + s.lruMu.RLock() + rp, cached := s.lruGet(longest.ID, now) + s.lruMu.RUnlock() + if cached { + if rp != nil { + out[txID] = rp + } else { + fallback = append(fallback, txID) + } + continue + } + longestTx[longest.ID] = txID + missing = append(missing, longest.ID) + } + sort.Ints(missing) + for start := 0; start < len(missing); start += resolvedPathBatchSize { + chunk := missing[start:min(start+resolvedPathBatchSize, len(missing))] + args := make([]interface{}, len(chunk)) + for i, id := range chunk { + args[i] = id + } + rows, err := s.db.conn.Query(`SELECT id, resolved_path FROM observations WHERE id IN (?`+strings.Repeat(",?", len(chunk)-1)+`)`, args...) + stored := make(map[int][]*string, len(chunk)) + if err == nil { + for rows.Next() { + var id int + var raw sql.NullString + if scanErr := rows.Scan(&id, &raw); scanErr != nil { + err = scanErr + break + } + if raw.Valid { + stored[id] = unmarshalResolvedPath(raw.String) + } + } + if err == nil { + err = rows.Err() + } + rows.Close() + } + for _, id := range chunk { + txID := longestTx[id] + if err != nil { + // Retain the scalar error/recovery behavior for a failed batch. + if rp := s.fetchResolvedPathForTxBestAt(txID, snapshots[txID], now); rp != nil { + out[txID] = rp + } + continue + } + rp, present := stored[id] + if present { + s.lruMu.Lock() + s.lruPut(id, rp) + s.lruMu.Unlock() + } + if rp != nil { + out[txID] = rp + } else { + fallback = append(fallback, txID) + } + } + } + s.loadFallbackResolvedPathsAt(fallback, snapshots, out, now) + return out +} + +func (s *PacketStore) loadFallbackResolvedPathsAt(txIDs []int, snapshots map[int][]*StoreObs, out map[int][]*string, now int64) { + sort.Ints(txIDs) + for start := 0; start < len(txIDs); start += resolvedPathBatchSize { + chunk := txIDs[start:min(start+resolvedPathBatchSize, len(txIDs))] + args := make([]interface{}, len(chunk)) + wanted := make(map[int]int) + for i, id := range chunk { + args[i] = id + for _, obs := range snapshots[id] { + wanted[obs.ID] = id + } + } + rows, err := s.db.conn.Query(`SELECT id, transmission_id, resolved_path FROM observations + WHERE resolved_path IS NOT NULL AND transmission_id IN (?`+strings.Repeat(",?", len(chunk)-1)+`)`, args...) + stored := make(map[int]map[int][]*string, len(chunk)) + if err == nil { + for rows.Next() { + var obsID, txID int + var raw sql.NullString + if scanErr := rows.Scan(&obsID, &txID, &raw); scanErr != nil { + err = scanErr + break + } + if !raw.Valid || raw.String == "" { + continue + } + // Historical rows absent from the immutable snapshot cannot + // win. Do not decode or retain their paths in this request. + if owner, ok := wanted[obsID]; !ok || owner != txID { + continue + } + if stored[txID] == nil { + stored[txID] = make(map[int][]*string) + } + stored[txID][obsID] = unmarshalResolvedPath(raw.String) + } + if err == nil { + err = rows.Err() + } + rows.Close() + } + for _, txID := range chunk { + if err != nil { + if rp := s.fetchResolvedPathForTxBestAt(txID, snapshots[txID], now); rp != nil { + out[txID] = rp + } + continue + } + obs := snapshots[txID] + rp, id := longestStoredResolvedPath(len(obs), func(i int) resolvedPathObservation { + return resolvedPathObservation{id: obs[i].ID, pathJSON: obs[i].PathJSON} + }, stored[txID]) + if rp != nil { + out[txID] = rp + if id != 0 { + s.lruMu.Lock() + s.lruPut(id, rp) + s.lruMu.Unlock() + } + } + } } - return best } // resolvedPubkeyHash computes a fast 64-bit hash for membership index keying. diff --git a/cmd/server/routes.go b/cmd/server/routes.go index de6850110..0afc5e55d 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -2454,13 +2454,12 @@ func (s *Server) handleNodePaths(w http.ResponseWriter, r *http.Request) { // Falls back to biased re-resolve only when a tx has no persisted // resolved_path (older data / async backfill incomplete); in that case // there's no canonical answer to be consistent with. - canonicalRP := make(map[int][]*string, len(candidates)) lruNow := s.store.lruNow() + canonicalSnapshots := make(map[int][]*StoreObs, len(candidates)) for _, tx := range candidates { - if rp := s.store.fetchResolvedPathForTxBestAt(tx.ID, observations[tx.ID], lruNow); rp != nil { - canonicalRP[tx.ID] = rp - } + canonicalSnapshots[tx.ID] = observations[tx.ID] } + canonicalRP := s.store.loadCanonicalResolvedPathsAt(canonicalSnapshots, lruNow) // Deferred exact-pubkey confirmation (see needsConfirm above): only for // hash-index candidates that have no canonical resolved_path, because