From 6c9d69cce8ce7dae158902e8ec379debf3a69960 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 24 Oct 2026 10:45:25 +0000 Subject: [PATCH] fix(paths): pin LRU insert order, share-safe longestTx, one lruMu hold per chunk (#548) Follow-ups from #542 for loadCanonicalResolvedPathsAt: - N1/N3: tests pin ascending chunk insert order and the survivors when one request inserts past lruMaxSize (highest ids); both sort.Ints calls covered. - N2: test that an absent row / NULL resolved_path never enters the LRU. - N4: longestTx is map[int][]int, so txs sharing a longest observation id are all answered. - N5: lruPutBatch takes lruMu once per chunk instead of once per id. Co-Authored-By: Claude Sonnet 5.5 --- cmd/server/paths_lru_548_test.go | 253 +++++++++++++++++++++++++++++++ cmd/server/resolved_index.go | 72 ++++++--- 2 files changed, 303 insertions(+), 22 deletions(-) create mode 100644 cmd/server/paths_lru_548_test.go diff --git a/cmd/server/paths_lru_548_test.go b/cmd/server/paths_lru_548_test.go new file mode 100644 index 000000000..da5c72184 --- /dev/null +++ b/cmd/server/paths_lru_548_test.go @@ -0,0 +1,253 @@ +package main + +import ( + "sort" + "sync" + "testing" + "time" +) + +// Follow-ups from #542 (issue #548) for loadCanonicalResolvedPathsAt. + +// lruOrderSnapshot returns lruOrder under the LRU lock. +func lruOrderSnapshot(s *PacketStore) []int { + s.lruMu.RLock() + defer s.lruMu.RUnlock() + return append([]int(nil), s.lruOrder...) +} + +// N3: one cold request inserts in ascending id order, chunk after chunk, for +// both the primary pass and the fallback pass. Removing either sort.Ints makes +// the order follow map iteration and fails here. +func TestCanonicalPaths548ChunkInsertOrderAscending(t *testing.T) { + for _, fallback := range []bool{false, true} { + name := "primary" + if fallback { + name = "fallback" + } + t.Run(name, func(t *testing.T) { + s := batch2224Store(t, false) + const n = 1200 // three chunks, below lruMaxSize + snapshots := seed2224(t, s, n, fallback) + got := s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + if len(got) != n { + t.Fatalf("got %d paths, want %d", len(got), n) + } + order := lruOrderSnapshot(s) + if len(order) != n { + t.Fatalf("lruOrder has %d entries, want %d", len(order), n) + } + if !sort.IntsAreSorted(order) { + t.Fatalf("insert order is not ascending by observation id: %v ...", order[:10]) + } + }) + } +} + +// N1: when one request inserts past lruMaxSize, the intended survivors are the +// highest observation ids (ascending insert, FIFO eviction drops the lowest). +func TestCanonicalPaths548EvictionOrderPastCapacity(t *testing.T) { + for _, fallback := range []bool{false, true} { + name := "primary" + if fallback { + name = "fallback" + } + t.Run(name, func(t *testing.T) { + s := batch2224Store(t, false) + const n = lruMaxSize + 50 + snapshots := seed2224(t, s, n, fallback) + // The cached observation is the primary (odd id 2*tx-1) or, on + // the fallback pass, its sibling (even id 2*tx). + cachedID := func(tx int) int { + if fallback { + return 2 * tx + } + return 2*tx - 1 + } + s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + s.lruMu.RLock() + defer s.lruMu.RUnlock() + if len(s.apiResolvedPathLRU) != lruMaxSize { + t.Fatalf("LRU holds %d entries, want %d", len(s.apiResolvedPathLRU), lruMaxSize) + } + for tx := 1; tx <= 50; tx++ { + if _, ok := s.apiResolvedPathLRU[cachedID(tx)]; ok { + t.Fatalf("lowest id %d survived; eviction order is not ascending", cachedID(tx)) + } + } + for tx := n - 49; tx <= n; tx++ { + if _, ok := s.apiResolvedPathLRU[cachedID(tx)]; !ok { + t.Fatalf("highest id %d was evicted", cachedID(tx)) + } + } + }) + } +} + +// N2: an absent row and a NULL resolved_path are not cached. The sibling that +// supplies the fallback is. +func TestCanonicalPaths548AbsentAndNullNotCached(t *testing.T) { + s := batch2224Store(t, false) + for _, row := range []struct { + id, tx int + path string + rp interface{} + }{ + {10, 1, `["aa","bb"]`, nil}, // longest, NULL resolved_path + {11, 1, `["cc"]`, `["sibling"]`}, // shorter sibling supplies fallback + {20, 2, `["aa"]`, `["present"]`}, // plain hit + } { + if _, err := s.db.conn.Exec(`INSERT INTO observations VALUES (?,?,?,?)`, row.id, row.tx, row.path, row.rp); err != nil { + t.Fatal(err) + } + } + snapshots := map[int][]*StoreObs{ + 1: {{ID: 10, PathJSON: `["aa","bb"]`}, {ID: 11, PathJSON: `["cc"]`}}, + 2: {{ID: 20, PathJSON: `["aa"]`}}, + 3: {{ID: 30, PathJSON: `["aa"]`}}, // row 30 does not exist + } + got := s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + if len(got[1]) != 1 || *got[1][0] != "sibling" || len(got[2]) != 1 || *got[2][0] != "present" { + t.Fatalf("unexpected paths: %+v", got) + } + if _, ok := got[3]; ok { + t.Fatal("absent row produced a path") + } + s.lruMu.RLock() + defer s.lruMu.RUnlock() + for _, id := range []int{10, 30} { + if _, ok := s.apiResolvedPathLRU[id]; ok { + t.Fatalf("observation %d (NULL/absent) entered the LRU", id) + } + } + for _, id := range []int{11, 20} { + if _, ok := s.apiResolvedPathLRU[id]; !ok { + t.Fatalf("observation %d should be cached", id) + } + } +} + +// N4: two transmissions whose longest observation has the same id must both +// be answered. A plain obsID->txID map keeps only the last one. +func TestCanonicalPaths548SharedLongestObservation(t *testing.T) { + s := batch2224Store(t, false) + if _, err := s.db.conn.Exec(`INSERT INTO observations VALUES (5,1,'["aa"]','["shared"]')`); err != nil { + t.Fatal(err) + } + snapshots := map[int][]*StoreObs{ + 1: {{ID: 5, PathJSON: `["aa"]`}}, + 2: {{ID: 5, PathJSON: `["aa"]`}}, + } + got := s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + for _, tx := range []int{1, 2} { + if len(got[tx]) != 1 || *got[tx][0] != "shared" { + t.Fatalf("tx %d lost its path: %+v", tx, got) + } + } +} + +// N5: lruMu is taken once per chunk, not once per id. The clock hook runs +// under the write lock inside lruPut; at the second put it starts a reader +// and holds the lock long enough for it to queue. The reader then sees the +// whole chunk if the chunk is written under one lock hold, and only the first +// two entries if the lock is released between ids. +func TestCanonicalPaths548LRULockedOncePerChunk(t *testing.T) { + for _, fallback := range []bool{false, true} { + name := "primary" + if fallback { + name = "fallback" + } + t.Run(name, func(t *testing.T) { + s := batch2224Store(t, false) + const n = 6 + snapshots := seed2224(t, s, n, fallback) + var calls int + var wg sync.WaitGroup + seen := -1 + s.lruClock = func() time.Time { + calls++ + if calls == 2 { + wg.Add(1) + go func() { + defer wg.Done() + s.lruMu.RLock() + seen = len(s.apiResolvedPathLRU) + s.lruMu.RUnlock() + }() + time.Sleep(50 * time.Millisecond) + } + return lruEpoch + } + s.loadCanonicalResolvedPathsAt(snapshots, 0) + wg.Wait() + if calls != n { + t.Fatalf("lruPut ran %d times, want %d", calls, n) + } + if seen != n { + t.Fatalf("reader saw %d entries mid-chunk, want all %d (lock released between ids)", seen, n) + } + }) + } +} + +// N3 (fallback pass): when the primaries are cached as empty, the fallback +// list is built in map iteration order, so only sort.Ints(txIDs) makes the +// fallback chunks and their LRU inserts ascending. +func TestCanonicalPaths548FallbackListSortedFromCachedEmpty(t *testing.T) { + s := batch2224Store(t, false) + const n = 1200 + snapshots := seed2224(t, s, n, true) + s.lruMu.Lock() + for tx := 1; tx <= n; tx++ { + s.lruPut(2*tx-1, nil) // primary cached as an empty path + } + s.lruMu.Unlock() + got := s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + if len(got) != n { + t.Fatalf("got %d paths, want %d", len(got), n) + } + order := lruOrderSnapshot(s) + if len(order) != 2*n { + t.Fatalf("lruOrder has %d entries, want %d", len(order), 2*n) + } + if tail := order[n:]; !sort.IntsAreSorted(tail) { + t.Fatalf("fallback inserts are not ascending: %v ...", tail[:10]) + } +} + +// BenchmarkCanonicalPaths548LRUContention measures the N5 change under reader +// pressure on lruMu: each lruMu write acquisition has to wait for in-flight +// readers, so per-id locking pays that once per id and per-chunk locking once +// per chunk. +func BenchmarkCanonicalPaths548LRUContention(b *testing.B) { + s := batch2224Store(b, false) + snapshots := seed2224(b, s, 5000, false) + stop := make(chan struct{}) + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + s.lruMu.RLock() + _, _ = s.lruGet(1, 0) + s.lruMu.RUnlock() + } + }() + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + resetLRU(s) + batch2224Sink = s.loadCanonicalResolvedPathsAt(snapshots, s.lruNow()) + } + b.StopTimer() + close(stop) + wg.Wait() + batch2224Sink = nil +} diff --git a/cmd/server/resolved_index.go b/cmd/server/resolved_index.go index 64535f821..abe62cb89 100644 --- a/cmd/server/resolved_index.go +++ b/cmd/server/resolved_index.go @@ -104,7 +104,9 @@ func (s *PacketStore) loadCanonicalResolvedPathsAt(snapshots map[int][]*StoreObs if s.db == nil || s.db.conn == nil { return out } - longestTx := make(map[int]int, len(snapshots)) + // obs id -> every tx whose longest observation it is; the same id can + // be the longest of several txs, and each must be answered. + longestTx := make(map[int][]int, len(snapshots)) missing := make([]int, 0, len(snapshots)) fallback := make([]int, 0) for txID, observations := range snapshots { @@ -128,9 +130,14 @@ func (s *PacketStore) loadCanonicalResolvedPathsAt(snapshots map[int][]*StoreObs } continue } - longestTx[longest.ID] = txID - missing = append(missing, longest.ID) + if _, queued := longestTx[longest.ID]; !queued { + missing = append(missing, longest.ID) + } + longestTx[longest.ID] = append(longestTx[longest.ID], txID) } + // Ascending ids make chunk membership and LRU insert order independent of + // map iteration: when one request inserts past lruMaxSize, FIFO eviction + // drops the lowest ids and the highest survive (#548). sort.Ints(missing) for start := 0; start < len(missing); start += resolvedPathBatchSize { chunk := missing[start:min(start+resolvedPathBatchSize, len(missing))] @@ -157,27 +164,32 @@ func (s *PacketStore) loadCanonicalResolvedPathsAt(snapshots map[int][]*StoreObs } rows.Close() } + putIDs := make([]int, 0, len(chunk)) + putRPs := make([][]*string, 0, len(chunk)) 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() + // Only a row that exists with a non-NULL resolved_path is cached; + // an absent row or NULL must stay a miss (#548). + if err == nil && present { + putIDs = append(putIDs, id) + putRPs = append(putRPs, rp) } - if rp != nil { - out[txID] = rp - } else { - fallback = append(fallback, txID) + for _, txID := range 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 + } + if rp != nil { + out[txID] = rp + } else { + fallback = append(fallback, txID) + } } } + s.lruPutBatch(putIDs, putRPs) } s.loadFallbackResolvedPathsAt(fallback, snapshots, out, now) return out @@ -224,6 +236,8 @@ func (s *PacketStore) loadFallbackResolvedPathsAt(txIDs []int, snapshots map[int } rows.Close() } + putIDs := make([]int, 0, len(chunk)) + putRPs := make([][]*string, 0, len(chunk)) for _, txID := range chunk { if err != nil { if rp := s.fetchResolvedPathForTxBestAt(txID, snapshots[txID], now); rp != nil { @@ -238,12 +252,12 @@ func (s *PacketStore) loadFallbackResolvedPathsAt(txIDs []int, snapshots map[int if rp != nil { out[txID] = rp if id != 0 { - s.lruMu.Lock() - s.lruPut(id, rp) - s.lruMu.Unlock() + putIDs = append(putIDs, id) + putRPs = append(putRPs, rp) } } } + s.lruPutBatch(putIDs, putRPs) } } @@ -635,6 +649,20 @@ func (s *PacketStore) lruPut(obsID int, rp []*string) { s.lruOrder = append(s.lruOrder, obsID) } +// lruPutBatch adds the entries in order under one lruMu hold, so a chunk costs +// one lock round-trip instead of one per id. It takes lruMu itself; callers +// must not hold it. +func (s *PacketStore) lruPutBatch(ids []int, rps [][]*string) { + if len(ids) == 0 { + return + } + s.lruMu.Lock() + for i, id := range ids { + s.lruPut(id, rps[i]) + } + s.lruMu.Unlock() +} + // lruDelete removes an entry. Must be called under s.lruMu write lock. func (s *PacketStore) lruDelete(obsID int) { if s.apiResolvedPathLRU == nil {