diff --git a/cmd/server/pathlen_fast_test.go b/cmd/server/pathlen_fast_test.go index fb182d56c..970e18141 100644 --- a/cmd/server/pathlen_fast_test.go +++ b/cmd/server/pathlen_fast_test.go @@ -9,8 +9,13 @@ import ( // pathLen is called once per observation of every candidate transmission in // /api/nodes/{pk}/paths and /hop_analytics. The fast path must agree with the // original json.Unmarshal-based implementation (pathLenSlow) for every input. +// +// The two EquivalenceGuard tests compare pathLen with the reference. That +// comparison alone also passes on a build without a fast path (pathLen == +// pathLenSlow), so each test also checks pathLenFast directly and requires it +// to serve inputs: a fast path that declines everything fails them (#277). -func TestPathLenFast_MatchesReference(t *testing.T) { +func TestPathLen_EquivalenceGuard_Corpus(t *testing.T) { cases := []string{ "", " ", "invalid", "null", "true", "0", `""`, `"aa"`, `{}`, `{"a":1}`, `[]`, ` [] `, "\t[\n]\r\n", `[ ]`, `[,]`, `[""]`, `["",""]`, @@ -27,28 +32,57 @@ func TestPathLenFast_MatchesReference(t *testing.T) { if got, want := pathLen(in), pathLenSlow(in); got != want { t.Errorf("pathLen(%q) = %d, want %d (reference)", in, got, want) } + if got, ok := pathLenFast(in); ok && got != pathLenSlow(in) { + t.Errorf("pathLenFast(%q) = %d, want %d (reference)", in, got, pathLenSlow(in)) + } + } + // The well-formed members of the corpus must be served by the fast path + // itself, not by the fallback. + for _, in := range []string{ + `[]`, ` [] `, "\t[\n]\r\n", `[ ]`, `[""]`, `["",""]`, `["aa"]`, `["aa","bb"]`, + `["aa","bb","cc"]`, ` [ "aa" , "bb" ] `, `["a,b","c]d","[e"]`, + `["aa","bb","cc","dd","ee","ff","00","11"]`, + } { + if _, ok := pathLenFast(in); !ok { + t.Errorf("pathLenFast(%q) fell back; the corpus's well-formed inputs must stay on the fast path", in) + } } } -// TestPathLenFast_RandomisedAgainstReference throws structurally "almost -// valid" JSON at both implementations. The token alphabet is chosen so most -// strings are close to the fast path's grammar, which is where a divergence -// would hide. -func TestPathLenFast_RandomisedAgainstReference(t *testing.T) { +// TestPathLen_EquivalenceGuard_Randomised throws structurally "almost valid" +// JSON at both implementations. The token alphabet is chosen so most strings +// are close to the fast path's grammar, which is where a divergence would +// hide. +func TestPathLen_EquivalenceGuard_Randomised(t *testing.T) { tokens := []string{ "[", "]", ",", `"`, `"aa"`, `"bb"`, `""`, " ", "\n", "\t", `\`, `\"`, "a", "1", "null", "é", "\x00", "{", "}", ":", "-", "e", } rng := rand.New(rand.NewSource(1352)) + fastNonEmpty := 0 for i := 0; i < 300000; i++ { var b strings.Builder for j, n := 0, rng.Intn(14); j < n; j++ { b.WriteString(tokens[rng.Intn(len(tokens))]) } in := b.String() - if got, want := pathLen(in), pathLenSlow(in); got != want { + want := pathLenSlow(in) + if got := pathLen(in); got != want { t.Fatalf("pathLen(%q) = %d, want %d (reference)", in, got, want) } + if got, ok := pathLenFast(in); ok { + if got != want { + t.Fatalf("pathLenFast(%q) = %d, want %d (reference)", in, got, want) + } + if got > 0 { + fastNonEmpty++ + } + } + } + // With this seed the fast path serves 11 non-empty arrays. Zero would mean + // the comparison above never exercised it. + if fastNonEmpty == 0 { + t.Error("pathLenFast served no non-empty input; the randomised comparison did not exercise the fast path") } } diff --git a/cmd/server/paths_confirm_deferred_test.go b/cmd/server/paths_confirm_deferred_test.go index 7428f5e98..0675405d5 100644 --- a/cmd/server/paths_confirm_deferred_test.go +++ b/cmd/server/paths_confirm_deferred_test.go @@ -59,6 +59,32 @@ func reloadConfirmStore(t *testing.T, srv *Server) *PacketStore { return store } +// lruHasTx reports whether any observation of txID has a resolved-path LRU +// entry, i.e. whether a handler read the tx's canonical path. +func lruHasTx(store *PacketStore, txID int) bool { + store.mu.RLock() + tx := store.byTxID[txID] + store.mu.RUnlock() + if tx == nil { + return false + } + store.lruMu.RLock() + defer store.lruMu.RUnlock() + for _, obs := range tx.Observations { + if _, ok := store.apiResolvedPathLRU[obs.ID]; ok { + return true + } + } + return false +} + +func resetLRU(store *PacketStore) { + store.lruMu.Lock() + defer store.lruMu.Unlock() + store.apiResolvedPathLRU = make(map[int]resolvedPathLRUEntry, lruMaxSize) + store.lruOrder = store.lruOrder[:0] +} + type confirmPathsResp struct { Paths []json.RawMessage `json:"paths"` TotalTransmissions int `json:"totalTransmissions"` @@ -94,7 +120,12 @@ func TestNodePaths_NoPerCandidateSQLConfirm(t *testing.T) { // contain the queried pubkey (hash collision, or an index entry made stale by // a later resolved_path overwrite) must still be excluded — now by the // canonical-path check instead of the SQL pre-filter. -func TestNodePaths_StaleIndexEntryStillExcluded(t *testing.T) { +// +// The exclusion assertions are regression guards: they held before #246 too, +// when the SQL pre-filter did the excluding. What tells before from after is +// how the decision is made: no confirmation query, and the tx's canonical path +// was read (it has an LRU entry; the SQL pre-filter dropped it unread). +func TestNodePaths_StaleIndexEntryExcludedWithoutSQLConfirm(t *testing.T) { srv, router := setupTestServer(t) staleID := seedConfirmTx(t, srv, "confirm_stale_hash", `["aa","bb"]`, `["aacafe0000000000","eeff00112233aabb"]`) @@ -113,6 +144,10 @@ func TestNodePaths_StaleIndexEntryStillExcluded(t *testing.T) { if strings.Contains(w.Body.String(), "confirm_stale_hash") { t.Error("tx whose resolved_path does not contain the target leaked into /paths via a stale index entry") } + if !lruHasTx(store, staleID) { + t.Error("/paths excluded the stale-index tx without reading its canonical resolved_path") + } + resetLRU(store) w = httptest.NewRecorder() router.ServeHTTP(w, httptest.NewRequest("GET", "/api/nodes/"+confirmTestTarget+"/hop_analytics", nil)) @@ -122,6 +157,9 @@ func TestNodePaths_StaleIndexEntryStillExcluded(t *testing.T) { if strings.Contains(w.Body.String(), "confirm_stale_hash") { t.Error("tx whose resolved_path does not contain the target leaked into /hop_analytics via a stale index entry") } + if !lruHasTx(store, staleID) { + t.Error("/hop_analytics excluded the stale-index tx without reading its canonical resolved_path") + } if q := store.confirmResolvedPathQueries.Load(); q != 0 { t.Errorf("confirmResolvedPathContains ran %d times, want 0", q) } @@ -130,7 +168,11 @@ func TestNodePaths_StaleIndexEntryStillExcluded(t *testing.T) { // Candidates with no canonical resolved_path are decided by the legacy // fallback arm, which still relies on the SQL confirmation. That behaviour is // preserved: the query runs (once) and a NULL resolved_path does not confirm. -func TestNodePaths_NoCanonicalPathStillConfirmedBySQL(t *testing.T) { +// +// Only the query count tells this apart from the code before #246, which ran +// the same query for every candidate: the exclusion assertion held then too, +// and the response is identical by design. The count is the behaviour pinned. +func TestNodePaths_NoCanonicalPathConfirmedBySQLExactlyOnce(t *testing.T) { srv, router := setupTestServer(t) nullID := seedConfirmTx(t, srv, "confirm_nullrp_hash", `["aa","bb"]`, "") store := reloadConfirmStore(t, srv) @@ -154,3 +196,40 @@ func TestNodePaths_NoCanonicalPathStillConfirmedBySQL(t *testing.T) { t.Errorf("confirmResolvedPathContains ran %d times, want exactly 1 (the tx without a canonical path)", q) } } + +// #277 point 2. Since #246 a candidate admitted by the hash index costs a +// canonical-path fetch, and an LRU entry, even when its stored path turns out +// not to contain the target. That stays cheap only because prefix collisions, +// the usual reason a /paths candidate does not belong to the node, are dropped +// by the index check before any fetch; only hash collisions and stale index +// entries get as far as the fetch. This pins that order for both endpoints. +func TestNodePaths_PrefixCollisionsExcludedBeforeCanonicalFetch(t *testing.T) { + srv, router := setupTestServer(t) + ownID := seedConfirmTx(t, srv, "collide_own", `["aa","bb"]`, `["`+confirmTestTarget+`","eeff00112233aabb"]`) + var otherIDs []int + for i := 0; i < 10; i++ { + // Same first-hop prefix as the target, resolved to another node. + otherIDs = append(otherIDs, seedConfirmTx(t, srv, fmt.Sprintf("collide_other_%02d", i), + `["aa","bb"]`, `["aacafe0000000000","eeff00112233aabb"]`)) + } + store := reloadConfirmStore(t, srv) + + for _, ep := range []string{"paths", "hop_analytics"} { + resetLRU(store) + body := nodeEndpointBody(t, router, ep) + if !strings.Contains(body, "collide_own") { + t.Errorf("%s: the target's own tx is missing", ep) + } + if !lruHasTx(store, ownID) { + t.Errorf("%s: the target's own tx was decided without its canonical path", ep) + } + for i, id := range otherIDs { + if strings.Contains(body, fmt.Sprintf("collide_other_%02d", i)) { + t.Errorf("%s: prefix-colliding tx %d attributed to the target", ep, id) + } + if lruHasTx(store, id) { + t.Errorf("%s: prefix-colliding tx %d reached the canonical-path fetch (LRU entry for a tx that is not the node's)", ep, id) + } + } + } +} diff --git a/cmd/server/paths_lru_staleness_test.go b/cmd/server/paths_lru_staleness_test.go new file mode 100644 index 000000000..7be3f0831 --- /dev/null +++ b/cmd/server/paths_lru_staleness_test.go @@ -0,0 +1,222 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +// #277: /paths and /hop_analytics decide membership from the canonical +// resolved_path (fetchResolvedPathForTxBest), which is served from +// apiResolvedPathLRU first. The ingestor's observation upsert +// (ON CONFLICT ... resolved_path = COALESCE(excluded.resolved_path, +// resolved_path)) can replace a stored path in place, keeping the row id, so +// the server's poll loop (WHERE o.id > ?) never sees the change. Nothing +// invalidated the LRU entry either, so the endpoints answered from the old +// path for as long as the entry survived FIFO eviction. + +// fakeLRUClock installs a controllable clock on the store's LRU and returns a +// function that moves it forward. +func fakeLRUClock(store *PacketStore) (advance func(time.Duration)) { + var now atomic.Int64 + now.Store(time.Now().UnixNano()) + store.lruClock = func() time.Time { return time.Unix(0, now.Load()) } + return func(d time.Duration) { now.Add(int64(d)) } +} + +// nodeEndpointBody GETs /api/nodes/{target}/{endpoint} and returns the body. +func nodeEndpointBody(t *testing.T, router http.Handler, endpoint string) string { + t.Helper() + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/api/nodes/"+confirmTestTarget+"/"+endpoint, nil)) + if w.Code != 200 { + t.Fatalf("%s: expected 200, got %d: %s", endpoint, w.Code, w.Body.String()) + } + return w.Body.String() +} + +// rewriteStoredResolvedPath does what the ingestor's upsert does when a +// re-delivered observation resolves differently: the row keeps its id and +// only resolved_path changes. +func rewriteStoredResolvedPath(t *testing.T, srv *Server, txID int, rpJSON string) { + t.Helper() + if _, err := srv.db.conn.Exec(`UPDATE observations SET resolved_path = ? WHERE transmission_id = ?`, rpJSON, txID); err != nil { + t.Fatalf("rewrite resolved_path: %v", err) + } +} + +func TestNodePaths_RewrittenResolvedPathNotServedFromStaleLRU(t *testing.T) { + srv, router := setupTestServer(t) + const hash = "lru_rewrite_hash" + txID := seedConfirmTx(t, srv, hash, `["aa","bb"]`, `["`+confirmTestTarget+`","eeff00112233aabb"]`) + store := reloadConfirmStore(t, srv) + advance := fakeLRUClock(store) + + // Warm the LRU: the stored path goes through the target. + for _, ep := range []string{"paths", "hop_analytics"} { + if !strings.Contains(nodeEndpointBody(t, router, ep), hash) { + t.Fatalf("%s: tx missing before the rewrite (setup)", ep) + } + } + + // The ingestor replaces the stored path with one that does not contain + // the target. The in-memory hash index still lists the tx under the + // target, so the canonical path is what decides. + rewriteStoredResolvedPath(t, srv, txID, `["aacafe0000000000","eeff00112233aabb"]`) + advance(time.Hour) + + for _, ep := range []string{"paths", "hop_analytics"} { + if strings.Contains(nodeEndpointBody(t, router, ep), hash) { + t.Errorf("%s: tx still attributed to the target an hour after its stored resolved_path stopped containing it (stale LRU entry)", ep) + } + } +} + +// The bound is a trade-off, not an invalidation: inside resolvedPathLRUTTL the +// cached path is still served without touching SQLite. This pins that the +// cache keeps doing its job (a TTL of 0, or a lookup that ignores the entry, +// would make every /paths candidate a primary-key read again). +func TestNodePaths_ResolvedPathLRUServesCachedEntryWithinTTL(t *testing.T) { + srv, router := setupTestServer(t) + const hash = "lru_within_ttl_hash" + txID := seedConfirmTx(t, srv, hash, `["aa","bb"]`, `["`+confirmTestTarget+`","eeff00112233aabb"]`) + store := reloadConfirmStore(t, srv) + advance := fakeLRUClock(store) + + if !strings.Contains(nodeEndpointBody(t, router, "paths"), hash) { + t.Fatal("tx missing before the rewrite (setup)") + } + rewriteStoredResolvedPath(t, srv, txID, `["aacafe0000000000","eeff00112233aabb"]`) + + // An entry must live long enough to absorb repeated dashboard requests. + advance(30 * time.Second) + if !strings.Contains(nodeEndpointBody(t, router, "paths"), hash) { + t.Fatal("cached resolved_path was re-read after 30 s; the LRU no longer saves repeat reads") + } + advance(resolvedPathLRUTTL - 31*time.Second) + if !strings.Contains(nodeEndpointBody(t, router, "paths"), hash) { + t.Error("cached resolved_path was re-read before resolvedPathLRUTTL elapsed") + } + advance(2 * time.Second) + if strings.Contains(nodeEndpointBody(t, router, "paths"), hash) { + t.Error("cached resolved_path still served after resolvedPathLRUTTL elapsed") + } +} + +func TestResolvedPathLRU_ExpiryAndInPlaceRefresh(t *testing.T) { + store := &PacketStore{} + store.initResolvedPathIndex() + advance := fakeLRUClock(store) + a, b := "aa", "bb" + + store.lruMu.Lock() + store.lruPut(7, []*string{&a}) + store.lruMu.Unlock() + if rp, ok := store.lruGet(7, store.lruNow()); !ok || *rp[0] != "aa" { + t.Fatalf("fresh entry not served: ok=%v", ok) + } + + advance(resolvedPathLRUTTL + time.Nanosecond) + if _, ok := store.lruGet(7, store.lruNow()); ok { + t.Fatal("entry older than resolvedPathLRUTTL served") + } + + // A re-read after expiry must replace the value and restart its age, + // without a second FIFO slot for the same id. + store.lruMu.Lock() + store.lruPut(7, []*string{&b}) + store.lruMu.Unlock() + if rp, ok := store.lruGet(7, store.lruNow()); !ok || *rp[0] != "bb" { + t.Fatalf("refreshed entry not served with the new value: ok=%v", ok) + } + if n := len(store.lruOrder); n != 1 { + t.Errorf("lruOrder has %d slots for one id, want 1", n) + } + + // /paths reads the clock once per request, so a lookup's now can be + // older than an entry stored meanwhile: that entry is fresh. + if _, ok := store.lruGet(7, store.lruNow()-int64(time.Second)); !ok { + t.Error("entry stored after the lookup's clock read was treated as expired") + } +} + +// Every staleness test installs lruClock, so this is the only test of the +// production clock. Its unit must be nanoseconds: lruGet compares ages with +// int64(resolvedPathLRUTTL), so a clock in ms or µs would stretch the 60 s +// TTL to days or hours while the hooked tests stay green. +func TestResolvedPathLRU_DefaultClockIsMonotonicNanoseconds(t *testing.T) { + store := &PacketStore{} + start := time.Now() + a := store.lruNow() + time.Sleep(time.Millisecond) + b := store.lruNow() + outer := time.Since(start) + if b <= a || b > int64(time.Hour) { + t.Errorf("lruNow went %d -> %d; want an increasing offset from lruEpoch", a, b) + } + // [a, b] lies inside [start, start+outer] on the same monotonic clock. + if d := b - a; d < int64(time.Millisecond) || d > int64(outer) { + t.Errorf("lruNow advanced %d across a 1 ms sleep (outer interval %d ns); want nanoseconds", d, outer.Nanoseconds()) + } +} + +// lruClock stands in for time.Now, so the hook and the default clock must +// measure in the same domain: installing the hook while entries exist must +// neither expire nor revive them. +func TestResolvedPathLRU_ClockHookSharesDefaultClockDomain(t *testing.T) { + store := &PacketStore{} + store.initResolvedPathIndex() + a := "aa" + + store.lruMu.Lock() + store.lruPut(7, []*string{&a}) // stored on the default clock + store.lruMu.Unlock() + + store.lruClock = time.Now + if _, ok := store.lruGet(7, store.lruNow()); !ok { + t.Fatal("entry stored on the default clock expired once lruClock = time.Now was installed") + } + store.lruClock = func() time.Time { return time.Now().Add(resolvedPathLRUTTL + time.Second) } + if _, ok := store.lruGet(7, store.lruNow()); ok { + t.Error("entry stored on the default clock still served after the hook moved past resolvedPathLRUTTL") + } +} + +// Entry ages cost a clock read, which is not free on every host (about 30 ns +// per read on a kvm-clock VM, against 15 ns for the whole cache hit). /paths +// and /hop_analytics look up one entry per candidate, so they read the clock +// once per request. A warm request with many candidates must read it once. +func TestNodePaths_WarmRequestReadsLRUClockOnce(t *testing.T) { + srv, router := setupTestServer(t) + for i := 0; i < 20; i++ { + seedConfirmTx(t, srv, fmt.Sprintf("lru_clock_%02d", i), `["aa","bb"]`, `["`+confirmTestTarget+`","eeff00112233aabb"]`) + } + store := reloadConfirmStore(t, srv) + var reads atomic.Int64 + base := time.Now() + store.lruClock = func() time.Time { reads.Add(1); return base } + + for _, ep := range []string{"paths", "hop_analytics"} { + nodeEndpointBody(t, router, ep) // warm + reads.Store(0) + body := nodeEndpointBody(t, router, ep) + // /paths groups identical paths (one sample hash per group), so + // count rows/occurrences rather than look for each hash. + if ep == "paths" { + var resp confirmPathsResp + if err := json.Unmarshal([]byte(body), &resp); err != nil || resp.TotalTransmissions < 20 { + t.Fatalf("paths: totalTransmissions = %d (err %v), want >= 20 (setup)", resp.TotalTransmissions, err) + } + } else if strings.Count(body, "lru_clock_") < 20 { + t.Fatalf("%s: seeded txs missing (setup)", ep) + } + if n := reads.Load(); n != 1 { + t.Errorf("%s: warm request read the LRU clock %d times for 20+ candidates, want 1", ep, n) + } + } +} diff --git a/cmd/server/resolved_index.go b/cmd/server/resolved_index.go index d2bcafe31..8dc76e97c 100644 --- a/cmd/server/resolved_index.go +++ b/cmd/server/resolved_index.go @@ -14,6 +14,7 @@ import ( "hash/fnv" "log" "strings" + "time" ) // resolvedPubkeyHash computes a fast 64-bit hash for membership index keying. @@ -200,17 +201,21 @@ func (s *PacketStore) fetchResolvedPathsForTx(txID int) map[int][]*string { // fetchResolvedPathForObs fetches resolved_path for a single observation, // using the LRU cache. func (s *PacketStore) fetchResolvedPathForObs(obsID int) []*string { + return s.fetchResolvedPathForObsAt(obsID, s.lruNow()) +} + +// fetchResolvedPathForObsAt is fetchResolvedPathForObs with the LRU clock +// read by the caller (lruNow), so a loop over many candidates reads it once. +func (s *PacketStore) fetchResolvedPathForObsAt(obsID int, now int64) []*string { if s.db == nil || s.db.conn == nil { return nil } // Check LRU cache first s.lruMu.RLock() - if s.apiResolvedPathLRU != nil { - if entry, ok := s.apiResolvedPathLRU[obsID]; ok { - s.lruMu.RUnlock() - return entry - } + if rp, ok := s.lruGet(obsID, now); ok { + s.lruMu.RUnlock() + return rp } s.lruMu.RUnlock() @@ -241,6 +246,14 @@ func (s *PacketStore) fetchResolvedPathForObs(obsID int) []*string { // #810 by checking all observations and falling back to the longest sibling // that has a stored path. func (s *PacketStore) fetchResolvedPathForTxBest(tx *StoreTx) []*string { + return s.fetchResolvedPathForTxBestAt(tx, s.lruNow()) +} + +// fetchResolvedPathForTxBestAt is fetchResolvedPathForTxBest with the LRU +// clock read by the caller. /paths and /hop_analytics call it once per +// candidate; reading the clock once per request keeps a cache hit as cheap as +// it was before entries had an age (#277). +func (s *PacketStore) fetchResolvedPathForTxBestAt(tx *StoreTx, now int64) []*string { if tx == nil || len(tx.Observations) == 0 { return nil } @@ -253,7 +266,7 @@ func (s *PacketStore) fetchResolvedPathForTxBest(tx *StoreTx) []*string { longestLen = l } } - if rp := s.fetchResolvedPathForObs(longest.ID); rp != nil { + if rp := s.fetchResolvedPathForObsAt(longest.ID, now); rp != nil { return rp } // Fallback: longest-path obs has no stored resolved_path. Query all @@ -291,12 +304,54 @@ func (s *PacketStore) fetchResolvedPathForTxBest(tx *StoreTx) []*string { const lruMaxSize = 10000 -// lruPut adds an entry. Must be called under s.lruMu write lock. +// resolvedPathLRUTTL bounds how long a cached resolved_path is served without +// re-reading the row (#277). The ingestor's observation upsert can replace a +// stored resolved_path in place (same row id), and the server's poll loop only +// reads new ids, so it cannot invalidate the entry when that happens. An +// expired entry is a cache miss: the next lookup costs one primary-key read. +const resolvedPathLRUTTL = 60 * time.Second + +type resolvedPathLRUEntry struct { + rp []*string + storedAt int64 // lruNow() at insert or refresh +} + +// lruEpoch anchors lruNow. time.Since reads the monotonic clock only, so +// entry ages are immune to wall-clock steps and cost one clock read. +var lruEpoch = time.Now() + +// lruNow is the LRU clock: nanoseconds since lruEpoch. The lruClock test hook +// stands in for time.Now and is measured from the same epoch, so installing it +// while entries exist keeps their ages. +func (s *PacketStore) lruNow() int64 { + if s.lruClock != nil { + return int64(s.lruClock().Sub(lruEpoch)) + } + return int64(time.Since(lruEpoch)) +} + +// lruGet returns the cached path unless the entry is older than +// resolvedPathLRUTTL at now. now may be read before the lookup (once per +// request), so an entry stored after it is fresh. Must be called under +// s.lruMu (read or write). +func (s *PacketStore) lruGet(obsID int, now int64) ([]*string, bool) { + e, ok := s.apiResolvedPathLRU[obsID] + if !ok || now-e.storedAt > int64(resolvedPathLRUTTL) { + return nil, false + } + return e.rp, true +} + +// lruPut adds an entry, or refreshes an existing one in place (keeping its +// FIFO slot, so lruOrder gets no duplicate). Must be called under s.lruMu +// write lock. func (s *PacketStore) lruPut(obsID int, rp []*string) { if s.apiResolvedPathLRU == nil { return } + entry := resolvedPathLRUEntry{rp: rp, storedAt: s.lruNow()} if _, exists := s.apiResolvedPathLRU[obsID]; exists { + s.apiResolvedPathLRU[obsID] = entry return } // Compact lruOrder if stale entries exceed 50% of capacity. @@ -322,7 +377,7 @@ func (s *PacketStore) lruPut(obsID int, rp []*string) { // stale entry — skip and continue } } - s.apiResolvedPathLRU[obsID] = rp + s.apiResolvedPathLRU[obsID] = entry s.lruOrder = append(s.lruOrder, obsID) } @@ -411,7 +466,7 @@ func (s *PacketStore) resolvedPubkeysForEvictionBatch(txIDs []int) map[int][]str func (s *PacketStore) initResolvedPathIndex() { s.resolvedPubkeyIndex = make(map[uint64][]int, 4096) s.resolvedPubkeyReverse = make(map[int][]uint64, 4096) - s.apiResolvedPathLRU = make(map[int][]*string, lruMaxSize) + s.apiResolvedPathLRU = make(map[int]resolvedPathLRUEntry, lruMaxSize) s.lruOrder = make([]int, 0, lruMaxSize) } diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 8294cc146..4e97b8d30 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -2291,8 +2291,9 @@ func (s *Server) handleNodePaths(w http.ResponseWriter, r *http.Request) { // 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() for _, tx := range candidates { - if rp := s.store.fetchResolvedPathForTxBest(tx); rp != nil { + if rp := s.store.fetchResolvedPathForTxBestAt(tx, lruNow); rp != nil { canonicalRP[tx.ID] = rp } } diff --git a/cmd/server/store.go b/cmd/server/store.go index 661d2f8b5..26f921d66 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -450,13 +450,14 @@ type PacketStore struct { // Resolved path membership index: xxhash → []txID (forward) and txID → []hashes (reverse). // Replaces per-StoreTx/StoreObs ResolvedPath []*string field (#800). - resolvedPubkeyIndex map[uint64][]int // hash(pubkey) → []txID - resolvedPubkeyReverse map[int][]uint64 // txID → []hashes indexed under - useResolvedPathIndex bool // feature flag (default true, off path = conservative) - maxResolvedPubkeyIndexEntries int // hard cap for size warning (0 = use default 5M) - apiResolvedPathLRU map[int][]*string // obsID → resolved path (LRU cache for API) - lruOrder []int // FIFO order for LRU eviction - lruMu sync.RWMutex // guards apiResolvedPathLRU + lruOrder + resolvedPubkeyIndex map[uint64][]int // hash(pubkey) → []txID + resolvedPubkeyReverse map[int][]uint64 // txID → []hashes indexed under + useResolvedPathIndex bool // feature flag (default true, off path = conservative) + maxResolvedPubkeyIndexEntries int // hard cap for size warning (0 = use default 5M) + apiResolvedPathLRU map[int]resolvedPathLRUEntry // obsID → resolved path (LRU cache for API, entries expire after resolvedPathLRUTTL) + lruOrder []int // FIFO order for LRU eviction + lruMu sync.RWMutex // guards apiResolvedPathLRU + lruOrder + lruClock func() time.Time // test hook standing in for time.Now in lruNow; nil = time.Since(lruEpoch), monotonic // confirmResolvedPathQueries counts SQL round-trips made by // confirmResolvedPathContains. Each one scans the tx's observation rows, @@ -10799,8 +10800,9 @@ func (s *PacketStore) GetNodeHopAnalytics(pubkey string, days int) (*NodeHopAnal } packets := make([]HopAnalyticsPacket, 0, len(confirmed)) + lruNow := s.lruNow() for _, tx := range confirmed { - rp := s.fetchResolvedPathForTxBest(tx) + rp := s.fetchResolvedPathForTxBestAt(tx, lruNow) if rp == nil { continue }