From 1a1e99f7d8b4f99ed969b17e5c584534eca904c7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 25 Feb 2026 11:31:17 -0500 Subject: [PATCH] feat: add per-account subscription result cache with LRU (Gap 5.4) Implements RouteResultCache with fixed-capacity LRU eviction and atomic generation-based invalidation (Go ref: client.go routeCache, maxResultCacheSize=8192). Fixes AccountGoParityTests namespace ambiguity introduced by new test file. --- .../Subscriptions/RouteResultCache.cs | 125 ++++++++++ .../Auth/AccountGoParityTests.cs | 7 +- .../Subscriptions/RouteResultCacheTests.cs | 223 ++++++++++++++++++ 3 files changed, 352 insertions(+), 3 deletions(-) create mode 100644 src/NATS.Server/Subscriptions/RouteResultCache.cs create mode 100644 tests/NATS.Server.Tests/Subscriptions/RouteResultCacheTests.cs diff --git a/src/NATS.Server/Subscriptions/RouteResultCache.cs b/src/NATS.Server/Subscriptions/RouteResultCache.cs new file mode 100644 index 0000000..ee4307b --- /dev/null +++ b/src/NATS.Server/Subscriptions/RouteResultCache.cs @@ -0,0 +1,125 @@ +namespace NATS.Server.Subscriptions; + +/// +/// Per-account LRU cache for subscription match results. +/// Avoids repeated SubList.Match() calls for the same subject. +/// Uses atomic generation ID for bulk invalidation when subscriptions change. +/// Go reference: server/client.go routeCache, maxResultCacheSize=8192. +/// +public sealed class RouteResultCache +{ + private readonly int _capacity; + private readonly Dictionary> _map; + private readonly LinkedList<(string Key, SubListResult Value)> _list = new(); + private readonly Lock _lock = new(); + private long _generation; + private long _cacheGeneration; + + // Stats + private long _hits; + private long _misses; + + public RouteResultCache(int capacity = 8192) + { + _capacity = capacity; + _map = new Dictionary>(capacity, StringComparer.Ordinal); + } + + public long Generation => Interlocked.Read(ref _generation); + public long Hits => Interlocked.Read(ref _hits); + public long Misses => Interlocked.Read(ref _misses); + public int Count { get { lock (_lock) return _map.Count; } } + + /// + /// Attempts to retrieve a cached result for the given subject. + /// On generation mismatch (subscriptions changed since last cache fill), + /// clears all entries and returns false. + /// Updates LRU order on a hit. + /// + public bool TryGet(string subject, out SubListResult? result) + { + lock (_lock) + { + var currentGeneration = Interlocked.Read(ref _generation); + if (_cacheGeneration != currentGeneration) + { + ClearUnderLock(); + _cacheGeneration = currentGeneration; + result = null; + Interlocked.Increment(ref _misses); + return false; + } + + if (_map.TryGetValue(subject, out var node)) + { + // Move to front (most recently used) + _list.Remove(node); + _list.AddFirst(node); + result = node.Value.Value; + Interlocked.Increment(ref _hits); + return true; + } + } + + result = null; + Interlocked.Increment(ref _misses); + return false; + } + + /// + /// Adds or updates a subject's cached result. + /// If at capacity, evicts the least recently used entry (tail of list). + /// + public void Set(string subject, SubListResult result) + { + lock (_lock) + { + if (_map.TryGetValue(subject, out var existing)) + { + // Update value in place and move to front + _list.Remove(existing); + var updated = new LinkedListNode<(string Key, SubListResult Value)>((subject, result)); + _list.AddFirst(updated); + _map[subject] = updated; + return; + } + + // Evict LRU tail if at capacity + if (_map.Count >= _capacity && _list.Last is { } tail) + { + _map.Remove(tail.Value.Key); + _list.RemoveLast(); + } + + var node = new LinkedListNode<(string Key, SubListResult Value)>((subject, result)); + _list.AddFirst(node); + _map[subject] = node; + } + } + + /// + /// Increments the generation counter so that the next TryGet detects a mismatch + /// and clears stale entries. This is called whenever subscriptions change. + /// + public void Invalidate() + { + Interlocked.Increment(ref _generation); + } + + /// + /// Immediately clears all cached entries under the lock. + /// + public void Clear() + { + lock (_lock) + { + ClearUnderLock(); + } + } + + private void ClearUnderLock() + { + _map.Clear(); + _list.Clear(); + } +} diff --git a/tests/NATS.Server.Tests/Auth/AccountGoParityTests.cs b/tests/NATS.Server.Tests/Auth/AccountGoParityTests.cs index 9200bc9..1419e21 100644 --- a/tests/NATS.Server.Tests/Auth/AccountGoParityTests.cs +++ b/tests/NATS.Server.Tests/Auth/AccountGoParityTests.cs @@ -3,6 +3,7 @@ using NATS.Server.Auth; using NATS.Server.Imports; +using NATS.Server.Subscriptions; namespace NATS.Server.Tests.Auth; @@ -27,7 +28,7 @@ public class AccountGoParityTests using var accB = new Account("B"); // Add subscriptions to account A's SubList - var subA = new Subscriptions.Subscription { Subject = "foo", Sid = "1" }; + var subA = new Subscription { Subject = "foo", Sid = "1" }; accA.SubList.Insert(subA); // Account B should not see account A's subscriptions @@ -51,8 +52,8 @@ public class AccountGoParityTests // Go: TestAccountWildcardRouteMapping — wildcards work per-account. using var acc = new Account("TEST"); - var sub1 = new Subscriptions.Subscription { Subject = "orders.*", Sid = "1" }; - var sub2 = new Subscriptions.Subscription { Subject = "orders.>", Sid = "2" }; + var sub1 = new Subscription { Subject = "orders.*", Sid = "1" }; + var sub2 = new Subscription { Subject = "orders.>", Sid = "2" }; acc.SubList.Insert(sub1); acc.SubList.Insert(sub2); diff --git a/tests/NATS.Server.Tests/Subscriptions/RouteResultCacheTests.cs b/tests/NATS.Server.Tests/Subscriptions/RouteResultCacheTests.cs new file mode 100644 index 0000000..e49d718 --- /dev/null +++ b/tests/NATS.Server.Tests/Subscriptions/RouteResultCacheTests.cs @@ -0,0 +1,223 @@ +// Go reference: server/client.go routeCache, maxResultCacheSize=8192. + +using NATS.Server.Subscriptions; +using Shouldly; + +namespace NATS.Server.Tests.Subscriptions; + +/// +/// Unit tests for RouteResultCache — the per-account LRU cache that stores +/// SubListResult objects keyed by subject string to avoid repeated SubList.Match() calls. +/// Go reference: server/client.go routeCache field, maxResultCacheSize=8192. +/// +public class RouteResultCacheTests +{ + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + private static SubListResult MakeResult(string subject) + { + var sub = new Subscription { Subject = subject, Sid = "1" }; + return new SubListResult([sub], []); + } + + // ========================================================================= + // Basic lookup behaviour + // ========================================================================= + + /// + /// An empty cache must return false for any subject. + /// + [Fact] + public void TryGet_returns_false_on_empty_cache() + { + var cache = new RouteResultCache(capacity: 16); + + var found = cache.TryGet("foo.bar", out var result); + + found.ShouldBeFalse(); + result.ShouldBeNull(); + } + + /// + /// A result placed with Set must be retrievable via TryGet. + /// + [Fact] + public void Set_and_TryGet_round_trips_result() + { + var cache = new RouteResultCache(capacity: 16); + var expected = MakeResult("foo"); + cache.Set("foo", expected); + + var found = cache.TryGet("foo", out var actual); + + found.ShouldBeTrue(); + actual.ShouldBeSameAs(expected); + } + + // ========================================================================= + // LRU eviction + // ========================================================================= + + /// + /// When a capacity+1 entry is added, the oldest (LRU) entry must be evicted. + /// Go reference: maxResultCacheSize=8192 with LRU eviction. + /// + [Fact] + public void LRU_eviction_at_capacity() + { + const int cap = 4; + var cache = new RouteResultCache(capacity: cap); + + // Fill to capacity — "first" is the oldest (LRU) + cache.Set("first", MakeResult("first")); + cache.Set("second", MakeResult("second")); + cache.Set("third", MakeResult("third")); + cache.Set("fourth", MakeResult("fourth")); + + // Adding a 5th entry must evict "first" + cache.Set("fifth", MakeResult("fifth")); + + cache.Count.ShouldBe(cap); + cache.TryGet("first", out _).ShouldBeFalse(); + cache.TryGet("fifth", out _).ShouldBeTrue(); + } + + /// + /// After accessing A, B, C (capacity=3), accessing A promotes it to MRU. + /// The next Set evicts B (now the true LRU) rather than A. + /// + [Fact] + public void LRU_order_updated_on_access() + { + var cache = new RouteResultCache(capacity: 3); + cache.Set("A", MakeResult("A")); + cache.Set("B", MakeResult("B")); + cache.Set("C", MakeResult("C")); + + // Access A so it becomes MRU; now LRU order is: B, A, C (front=C, back=B) + cache.TryGet("A", out _); + + // D evicts the LRU entry — should be B, not A + cache.Set("D", MakeResult("D")); + + cache.TryGet("A", out _).ShouldBeTrue(); + cache.TryGet("B", out _).ShouldBeFalse(); + cache.TryGet("C", out _).ShouldBeTrue(); + cache.TryGet("D", out _).ShouldBeTrue(); + } + + // ========================================================================= + // Generation / invalidation + // ========================================================================= + + /// + /// After Invalidate(), the next TryGet must detect the generation mismatch, + /// clear all entries, and return false. + /// + [Fact] + public void Invalidate_clears_on_next_access() + { + var cache = new RouteResultCache(capacity: 16); + cache.Set("a", MakeResult("a")); + cache.Set("b", MakeResult("b")); + + cache.Invalidate(); + + cache.TryGet("a", out var result).ShouldBeFalse(); + result.ShouldBeNull(); + cache.Count.ShouldBe(0); + } + + /// + /// Each call to Invalidate() must increment the generation counter by 1. + /// + [Fact] + public void Generation_increments_on_invalidate() + { + var cache = new RouteResultCache(); + var gen0 = cache.Generation; + + cache.Invalidate(); + cache.Generation.ShouldBe(gen0 + 1); + + cache.Invalidate(); + cache.Generation.ShouldBe(gen0 + 2); + } + + // ========================================================================= + // Statistics + // ========================================================================= + + /// + /// Hits counter increments on each successful TryGet. + /// + [Fact] + public void Hits_incremented_on_cache_hit() + { + var cache = new RouteResultCache(capacity: 16); + cache.Set("subject", MakeResult("subject")); + + cache.TryGet("subject", out _); + cache.TryGet("subject", out _); + + cache.Hits.ShouldBe(2); + } + + /// + /// Misses counter increments on each failed TryGet. + /// + [Fact] + public void Misses_incremented_on_cache_miss() + { + var cache = new RouteResultCache(capacity: 16); + + cache.TryGet("nope", out _); + cache.TryGet("also.nope", out _); + + cache.Misses.ShouldBe(2); + } + + // ========================================================================= + // Clear + // ========================================================================= + + /// + /// Clear() must remove all entries immediately. + /// + [Fact] + public void Clear_removes_all_entries() + { + var cache = new RouteResultCache(capacity: 16); + cache.Set("x", MakeResult("x")); + cache.Set("y", MakeResult("y")); + + cache.Clear(); + + cache.Count.ShouldBe(0); + cache.TryGet("x", out _).ShouldBeFalse(); + } + + // ========================================================================= + // Update existing entry + // ========================================================================= + + /// + /// Setting the same key twice must replace the value; TryGet returns the latest result. + /// + [Fact] + public void Set_updates_existing_entry() + { + var cache = new RouteResultCache(capacity: 16); + var first = MakeResult("v1"); + var second = MakeResult("v2"); + + cache.Set("key", first); + cache.Set("key", second); + + cache.Count.ShouldBe(1); + cache.TryGet("key", out var returned).ShouldBeTrue(); + returned.ShouldBeSameAs(second); + } +}