From 1c11ea9140fbd5422206c9b1c760f690bd5d7529 Mon Sep 17 00:00:00 2001 From: mgravell Date: Thu, 12 Nov 2026 06:43:38 +0000 Subject: [PATCH 1/3] spike: cache hits reuse the deserialized result for provably-shareable types An Entry memoizes ONE parsed result with the handler that produced it (first writer wins, handler-identity keyed, carried across slab compaction); the hit funnel serves the memo before touching the reader. Core auto-enrolls strings and no-reference value types (IsReferenceOrContainsReferences, conservative down-level); richer rules opt in through the new public IRespResultReuse on the handler - the HybridCache integration seam. --- .../Caching/RespClientCache.cs | 136 +++++++++- .../Contexts/RespExecutor.cs | 46 +++- .../Protocol/IRespResultReuse.cs | 34 +++ .../PublicAPI/PublicAPI.Unshipped.txt | 2 + .../RespCacheResultReuseTests.cs | 234 ++++++++++++++++++ 5 files changed, 447 insertions(+), 5 deletions(-) create mode 100644 src/StackExchange.Redis/Protocol/IRespResultReuse.cs create mode 100644 tests/StackExchange.Redis.Tests/RespCacheResultReuseTests.cs diff --git a/src/StackExchange.Redis/Caching/RespClientCache.cs b/src/StackExchange.Redis/Caching/RespClientCache.cs index 586b261ea..5f152519a 100644 --- a/src/StackExchange.Redis/Caching/RespClientCache.cs +++ b/src/StackExchange.Redis/Caching/RespClientCache.cs @@ -89,6 +89,7 @@ internal sealed partial class RespClientCache : IDisposable private long _expired; private long _refreshes; private long _servedStale; + private long _reused; /// Create a cache. /// How the cache is built; when null. @@ -311,6 +312,14 @@ private static bool IsReadOnly(CommandFlags flags) /// public long Refreshes => Volatile.Read(ref _refreshes); + /// Hits answered from a memoized parse, without touching the payload. + /// + /// Each of these is a deserialization that did not happen: the entry had already been parsed by + /// the same handler, the result was declared shareable, and the earlier result was handed out + /// again. See . + /// + public long Reused => Volatile.Read(ref _reused); + /// Reads answered from an entry the server had already invalidated. /// /// Its own counter rather than folded into hits: these are answers that were knowingly out of date, @@ -682,8 +691,28 @@ public bool TryGet( long maxAgeTicks, [NotNullWhen(true)] out RespPayload? payload, out bool shouldRefresh) + => TryGet(in frame, database, maxAgeTicks, out payload, out shouldRefresh, out _); + + /// + /// The rendered request. + /// The database the request ran against. + /// The caller's own freshness requirement. + /// The cached reply, retained. + /// Whether this caller claimed the refresh; see the other overload. + /// + /// The served entry's memoized-result slot, through which the caller may skip parsing entirely - + /// or record its parse for the next hit to skip. Empty on a miss, and harmlessly inert then. + /// + public bool TryGet( + in RespRequest frame, + int database, + long maxAgeTicks, + [NotNullWhen(true)] out RespPayload? payload, + out bool shouldRefresh, + out ResultReuse reuse) { shouldRefresh = false; + reuse = default; if (!_entries.TryGetValue(new EntryKey(frame, database), out var entry)) { @@ -693,7 +722,16 @@ public bool TryGet( // invalidated, but perhaps still servable for a moment - see TryServeStale for why that is a // decision rather than a shortcut - if (!entry.IsValid) return TryServeStale(entry, maxAgeTicks, out payload, out shouldRefresh); + if (!entry.IsValid) + { + if (TryServeStale(entry, maxAgeTicks, out payload, out shouldRefresh)) + { + reuse = new ResultReuse(this, entry); + return true; + } + + return false; + } if (entry.Payload.TryRetain()) { @@ -719,6 +757,7 @@ public bool TryGet( // line is written once per eviction cycle, not once per hit - see Entry.Touch entry.Touch(); payload = entry.Payload; + reuse = new ResultReuse(this, entry); return true; } @@ -765,7 +804,7 @@ public bool TryGet( /// /// /// - private bool TryServeStale(Entry entry, long maxAgeTicks, out RespPayload? payload, out bool shouldRefresh) + private bool TryServeStale(Entry entry, long maxAgeTicks, [NotNullWhen(true)] out RespPayload? payload, out bool shouldRefresh) { shouldRefresh = false; payload = null; @@ -1701,8 +1740,93 @@ internal void Publish() } } + /// + /// A served entry's memoized-result slot: the way a hit can skip deserialization entirely. + /// + /// + /// + /// The cache stores the raw reply, so every hit normally re-parses it - correct, but wasted work + /// when the parse always produces the same answer and that answer is safe to hand to more than one + /// caller. This slot pairs the entry with one parsed result and the handler that produced + /// it; a later hit with the same handler instance takes the result without touching the + /// payload. + /// + /// + /// Keyed by handler identity, not by result type. The same frame can legitimately be read + /// through different deserializers, and handing one handler an object another produced is type + /// confusion wearing a performance hat. Reference equality makes the check one compare, and the + /// handlers that matter - the inbuilt ones, a serializer per type - are singletons anyway. + /// + /// + /// Whether a result is safe to share is the caller's decision, made before ; + /// this type only stores and serves. First writer wins: one slot per entry, so a key read through + /// two handlers memoizes whichever parsed first and re-parses for the other - a bounded loss, where + /// a last-writer slot would let two hot readers evict each other forever. + /// + /// + /// Lifetime is the entry's, by construction: invalidation, expiry and eviction all discard the + /// entry, and the memo goes with it. A slab compaction carries it forward + /// (), because the copied bytes parse to the same answer. + /// + /// + /// The memo is not charged to : the budget meters bytes + /// the cache stores, and the managed size of an arbitrary parsed object is not knowable. For a + /// memoized string the entry's true footprint roughly doubles; a budget tuned to the knife's edge + /// should allow for that. + /// + /// + internal readonly struct ResultReuse + { + private readonly RespClientCache? _cache; + + // typed object because the ctor is internal and Entry is private: this struct can SEE Entry + // (nested types see their container's privates), but cannot NAME it in an internal signature + private readonly object? _entry; + + internal ResultReuse(RespClientCache cache, object entry) + { + _cache = cache; + _entry = entry; + } + + /// Take the memoized result, if this handler is the one that produced it. + /// What parsing the reply produces. + public bool TryGet(Protocol.IRespHandler handler, out TResult value) + { + if (_entry is Entry entry && entry.Memo is { } memo && ReferenceEquals(memo.Handler, handler)) + { + value = (TResult)memo.Value!; + Interlocked.Increment(ref _cache!._reused); + return true; + } + + value = default!; + return false; + } + + /// Offer a parsed result for later hits; kept only if the slot is still empty. + /// What parsing the reply produces. + public void TrySet(Protocol.IRespHandler handler, TResult value) + => (_entry as Entry)?.TrySetMemo(new Memo(handler, value)); + } + + /// A parsed result and the handler whose parse it is; see . + private sealed class Memo(object handler, object? value) + { + internal object Handler { get; } = handler; + internal object? Value { get; } = value; + } + private sealed class Entry(RespPayload payload, Dependency[] dependencies, RespRequest key, Slab? slab = null) { + /// The memoized parse of this entry's payload, if anybody stored one; see . + private Memo? _memo; + + internal Memo? Memo => Volatile.Read(ref _memo); + + /// First writer wins; see for why. + internal void TrySetMemo(Memo memo) => Interlocked.CompareExchange(ref _memo, memo, null); + // A single dependency - every single-key command, which is most of them - is held inline, and an array // only for more. Every hit asks IsValid, and the array was one more dependent load between the entry and // the generation it compares (and one more mid-life object per entry). The fill's own array dies young. @@ -1775,7 +1899,13 @@ internal bool TryClearReferenced() /// there. The referenced flag starts cleared. /// internal Entry CopyForward(RespPayload payload, RespRequest key, Slab? slab) - => new(payload, _many ?? [_single], key, slab) { FilledAt = FilledAt }; + { + var moved = new Entry(payload, _many ?? [_single], key, slab) { FilledAt = FilledAt }; + + // the copied bytes parse to the same answer, so the parse survives the move + moved._memo = Volatile.Read(ref _memo); + return moved; + } internal bool IsValid => _many is null ? _single.IsValid : Dependency.AllValid(_many); diff --git a/src/StackExchange.Redis/Contexts/RespExecutor.cs b/src/StackExchange.Redis/Contexts/RespExecutor.cs index 032407b61..393f055ab 100644 --- a/src/StackExchange.Redis/Contexts/RespExecutor.cs +++ b/src/StackExchange.Redis/Contexts/RespExecutor.cs @@ -1535,7 +1535,7 @@ private static bool TryServeFromCache( CommandFlags flags, [MaybeNullWhen(false)] out TResult result) { - if (!cache.TryGet(request.AsLookupKey(), executor.Database, maxAgeTicks, out var hit, out var refresh)) + if (!cache.TryGet(request.AsLookupKey(), executor.Database, maxAgeTicks, out var hit, out var refresh, out var reuse)) { result = default; return false; @@ -1556,7 +1556,15 @@ private static bool TryServeFromCache( try { - result = Parse(handler, hit); + // a memoized parse of this very payload - same handler, same bytes, so the same answer - + // served without touching the reader; else parse, and offer the result to the NEXT hit if + // the rules say it is safe to share + if (!reuse.TryGet(handler, out result)) + { + result = Parse(handler, hit); + if (MayShareResult(handler, in result)) reuse.TrySet(handler, result); + } + return true; } finally @@ -1565,6 +1573,40 @@ private static bool TryServeFromCache( } } + /// Whether a parsed result may be handed to more than one caller. + /// + /// + /// The core enrolls only what it can prove. A is immutable, and a value + /// type without inner references travels by copy - the memo holds the box, every caller unboxes + /// their own copy, and nothing one caller does is visible to another. Everything else - arrays, + /// mutable objects, structs that carry references ( among them) - is only + /// shared when the handler says so: it is the deserializer, so it is the one that knows whether + /// what it builds is immutable. That is the + /// opt-in, asked per value so the answer can depend on the runtime type, not just the declared one. + /// + /// + /// Down-level the reference detection does not exist, so the fallback enrolls only primitives and + /// enums: the mistakes are not the same size - a skipped share costs a re-parse, a wrong one hands + /// two callers the same mutable object. The same split as ReadOnlyLease.ClearOnReturn, with + /// the error direction flipped. + /// + /// + private static bool MayShareResult(IRespHandler handler, in TResult result) + => ResultSharing.Always + || (handler is Protocol.IRespResultReuse reuse && result is not null && reuse.MayReuse(in result)); + + /// The per-type half of , answered once per . + private static class ResultSharing + { + public static readonly bool Always = + typeof(TResult) == typeof(string) +#if NET + || !System.Runtime.CompilerServices.RuntimeHelpers.IsReferenceOrContainsReferences(); +#else + || typeof(TResult).IsPrimitive || typeof(TResult).IsEnum; +#endif + } + /// /// Re-fetch an ageing entry in the background, while its old value is still being served. /// diff --git a/src/StackExchange.Redis/Protocol/IRespResultReuse.cs b/src/StackExchange.Redis/Protocol/IRespResultReuse.cs new file mode 100644 index 000000000..ece311451 --- /dev/null +++ b/src/StackExchange.Redis/Protocol/IRespResultReuse.cs @@ -0,0 +1,34 @@ +namespace StackExchange.Redis.Protocol +{ + /// + /// Additionally implemented by an whose parsed results may be + /// handed to more than one caller, so a client-side cache hit can skip deserialization and serve the + /// result it already has. + /// + /// What parsing the reply produces. + /// + /// + /// The cache stores the raw reply; without this, every hit re-parses it. The core shares automatically + /// only what it can prove safe - strings, and value types without inner references - because sharing a + /// mutable object between callers is corruption, not caching. A handler that knows its results + /// are safe - immutable types, records with no settable state, a deserializer applying rules such as + /// the HybridCache immutability conventions - says so here, and its results are memoized too. + /// + /// + /// Asked per value rather than per type, so the answer can depend on what was actually produced: a + /// polymorphic declared type whose runtime type decides, or content the deserializer only trusts in + /// some shapes. An implementation with a purely type-level rule simply returns a constant. + /// + /// + /// Reuse is keyed by handler instance: a memoized result is only served back through the same + /// handler object that produced it, so implementations should be singletons (as the inbuilt handlers + /// are) for the memoization to pay. + /// + /// + public interface IRespResultReuse + { + /// Whether this value may be handed to more than one caller. + /// The value just parsed. + bool MayReuse(in TResult value); + } +} diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 1910e4231..bd1a18e6b 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -260,6 +260,8 @@ StackExchange.Redis.Protocol.IRespFormattableArgument StackExchange.Redis.Protocol.IRespFormattableArgument.WriteTo(scoped ref StackExchange.Redis.Protocol.RespRequestBuilder handler, string? format) -> void StackExchange.Redis.Protocol.IRespHandler StackExchange.Redis.Protocol.IRespHandler.Parse(ref RESPite.Messages.RespReader reader) -> TResult +StackExchange.Redis.Protocol.IRespResultReuse +StackExchange.Redis.Protocol.IRespResultReuse.MayReuse(in TResult value) -> bool StackExchange.Redis.Protocol.KeyRange StackExchange.Redis.Protocol.KeyRange.KeyRange() -> void StackExchange.Redis.Protocol.KeyRange.KeyRange(int offset, int length) -> void diff --git a/tests/StackExchange.Redis.Tests/RespCacheResultReuseTests.cs b/tests/StackExchange.Redis.Tests/RespCacheResultReuseTests.cs new file mode 100644 index 000000000..a3a1f0b20 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RespCacheResultReuseTests.cs @@ -0,0 +1,234 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using RESPite.Messages; +using StackExchange.Redis.Caching; +using StackExchange.Redis.Protocol; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Deserialization reuse on cache hits: a result the rules declare shareable is memoized on the entry, +/// and later hits are served the parsed result itself rather than a fresh parse of the payload. +/// +/// +/// +/// The sequence every test leans on: the FIRST read is the fill (its parse happens on the fill path and +/// is not memoized), the SECOND read is the first hit (it parses, and offers the result to the slot), and +/// the THIRD read onward is served from the memo. So "same instance" is asserted between the second and +/// third reads, and cache.Reused counts reads from the third on. +/// +/// +/// What the core auto-enrolls is deliberately narrow - strings and no-reference value types - and the +/// richer rules belong to the handler via ; these tests pin both +/// the enrollments and the refusals. +/// +/// +public class RespCacheResultReuseTests +{ + private const CommandFlags Readable = CommandFlags.CommandRetryReadOnly; + + private static (RespDatabaseContext Context, RespClientCache Cache, FakeExecutor Executor) Create(params string[] replies) + { + var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss }); + var executor = new FakeExecutor(replies); + var context = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache)); + return (context, cache, executor); + } + + [Fact] + public async Task AStringIsHandedBackAsTheSameInstance() + { + var (context, cache, executor) = Create("$5\r\nhello\r\n"); + using (cache) + { + var filled = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + var parsed = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + var reused = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + + Assert.Equal("hello", filled); + Assert.Equal("hello", parsed); + Assert.Same(parsed, reused); // the memo, not a re-decode + Assert.Equal(1, executor.Sends); + Assert.Equal(1L, cache.Reused); + } + } + + [Fact] + public async Task ANoReferenceStructIsReused() + { + var (context, cache, executor) = Create(":42\r\n"); + using (cache) + { + for (int i = 0; i < 4; i++) + { + Assert.Equal(42, await context.SendAsync($"{RedisCommand.STRLEN}{(RedisKey)"k"}", Readable)); + } + + Assert.Equal(1, executor.Sends); + + // fill, first hit (parses + memoizes), then two served from the box; a primitive qualifies + // under both the IsReferenceOrContainsReferences detection and the down-level fallback + Assert.Equal(2L, cache.Reused); + } + } + + [Fact] + public async Task ARedisValueIsNotAutoEnrolled() + { + // RedisValue is a struct that CARRIES a reference, so handing out the same box could hand out + // shared mutable state (the object it wraps); it must re-parse on every hit + var (context, cache, executor) = Create("$5\r\nhello\r\n"); + using (cache) + { + for (int i = 0; i < 3; i++) + { + Assert.Equal("hello", (string?)await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable)); + } + + Assert.Equal(1, executor.Sends); // still served from cache... + Assert.Equal(0L, cache.Reused); // ...but parsed each time + } + } + + [Fact] + public async Task AHandlerCanDeclareItsResultsShareable() + { + var (context, cache, executor) = Create("$5\r\nhello\r\n"); + using (cache) + { + var filled = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, ReusablePocoHandler.Instance); + var parsed = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, ReusablePocoHandler.Instance); + var reused = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, ReusablePocoHandler.Instance); + + Assert.Equal(5, filled.Length); + Assert.Same(parsed, reused); // a mutable-looking class, shared because the handler vouched for it + Assert.NotSame(filled, parsed); // the fill's own parse is not memoized; the first HIT's is + Assert.Equal(1, executor.Sends); + Assert.Equal(1L, cache.Reused); + } + } + + [Fact] + public async Task AHandlerThatDoesNotOptInReparsesEveryHit() + { + var (context, cache, executor) = Create("$5\r\nhello\r\n"); + using (cache) + { + var first = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, PlainPocoHandler.Instance); + var second = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, PlainPocoHandler.Instance); + var third = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, PlainPocoHandler.Instance); + + Assert.NotSame(second, third); + Assert.Equal(1, executor.Sends); + Assert.Equal(0L, cache.Reused); + } + } + + [Fact] + public async Task AHandlerCanRefusePerValue() + { + // the opt-in is asked per value, so a handler can trust some shapes and not others; this one + // refuses everything at runtime despite implementing the interface + var (context, cache, executor) = Create("$5\r\nhello\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, SelectivePocoHandler.Instance); + var second = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, SelectivePocoHandler.Instance); + var third = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, SelectivePocoHandler.Instance); + + Assert.NotSame(second, third); + Assert.Equal(0L, cache.Reused); + + // consulted on both HIT parses; the fill's parse is never memoized, so it is never asked + Assert.Equal(2, SelectivePocoHandler.Instance.Asked); + } + } + + [Fact] + public async Task TheMemoIsKeyedByHandlerInstance() + { + // two handlers, same frame, same TResult: the memo one handler stored must never be served + // through the other, because "same type" does not mean "same parse" + var (context, cache, executor) = Create("$5\r\nhello\r\n"); + using (cache) + { + var plain = new ScalingHandler(1); + var scaled = new ScalingHandler(10); + + for (int i = 0; i < 3; i++) + { + Assert.Equal(5, await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, plain)); + } + + // a different handler on the same cached entry: it must parse for itself, not inherit 5 + Assert.Equal(50, await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, scaled)); + Assert.Equal(50, await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, scaled)); + + Assert.Equal(1, executor.Sends); + } + } + + [Fact] + public async Task ALocalWriteDropsTheMemoWithTheEntry() + { + var (context, cache, executor) = Create("$1\r\na\r\n", "+OK\r\n", "$1\r\nb\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + var before = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); // memoized + Assert.Equal("a", before); + + // our own write: the entry - memo and all - must never be served again + _ = await context.SendAsync($"{RedisCommand.SET}{(RedisKey)"k"}{(RedisValue)"b"}", CommandFlags.None); + + var after = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + Assert.Equal("b", after); + Assert.Equal(3, executor.Sends); + } + } + + private sealed class Poco(int length) + { + public int Length => length; + } + + private sealed class ReusablePocoHandler : IRespHandler, IRespResultReuse + { + public static readonly ReusablePocoHandler Instance = new(); + + public Poco Parse(ref RespReader reader) => new(reader.ScalarLength()); + + public bool MayReuse(in Poco value) => true; + } + + private sealed class PlainPocoHandler : IRespHandler + { + public static readonly PlainPocoHandler Instance = new(); + + public Poco Parse(ref RespReader reader) => new(reader.ScalarLength()); + } + + private sealed class SelectivePocoHandler : IRespHandler, IRespResultReuse + { + public static readonly SelectivePocoHandler Instance = new(); + + private int _asked; + + public int Asked => Volatile.Read(ref _asked); + + public Poco Parse(ref RespReader reader) => new(reader.ScalarLength()); + + public bool MayReuse(in Poco value) + { + Interlocked.Increment(ref _asked); + return false; + } + } + + private sealed class ScalingHandler(int factor) : IRespHandler + { + public int Parse(ref RespReader reader) => reader.ScalarLength() * factor; + } +} From d19da6dc0dd0e21b760bbb81e959d40838994d7d Mon Sep 17 00:00:00 2001 From: mgravell Date: Thu, 12 Nov 2026 07:07:25 +0000 Subject: [PATCH 2/3] A memoized result charges the cache budget: IRespResultReuse gains SizeOf, and the core sizes what it enrolls The charge is paired with the Memo object itself - set wins charge, whoever takes the memo (removal, replacement, slab move via ownership transfer, with a tombstone closing the slot) credits exactly what was charged - so MaxBytes cannot drift under the discard/late-set race. Strings are header+UTF-16, boxed no-ref structs are header+Unsafe.SizeOf; handler-opted results volunteer an estimate, 0 meaning uncounted. --- .../Caching/RespClientCache.cs | 66 ++++++++++++++--- .../Contexts/RespExecutor.cs | 28 ++++++- .../Protocol/IRespResultReuse.cs | 12 +++ .../PublicAPI/PublicAPI.Unshipped.txt | 1 + .../RespCacheResultReuseTests.cs | 73 +++++++++++++++++++ 5 files changed, 169 insertions(+), 11 deletions(-) diff --git a/src/StackExchange.Redis/Caching/RespClientCache.cs b/src/StackExchange.Redis/Caching/RespClientCache.cs index 5f152519a..b4f63583a 100644 --- a/src/StackExchange.Redis/Caching/RespClientCache.cs +++ b/src/StackExchange.Redis/Caching/RespClientCache.cs @@ -1215,6 +1215,7 @@ private bool TryCompleteCore(in RespFill fill, RespPayload response) && _entries.TryUpdate(entryKey, entry, previous)) { Interlocked.Add(ref _bytes, entry.Bytes - previous.Bytes); + if (previous.TakeMemo() is { Bytes: not 0 } memo) Interlocked.Add(ref _bytes, -memo.Bytes); previous.Payload.Dispose(); // the superseded reply if (previous.Slab is { } superseded) ReleaseSlab(superseded); fill.Key.Dispose(); @@ -1324,6 +1325,7 @@ private void Release(Entry entry) { Interlocked.Decrement(ref _count); Interlocked.Add(ref _bytes, -entry.Bytes); + if (entry.TakeMemo() is { Bytes: not 0 } memo) Interlocked.Add(ref _bytes, -memo.Bytes); entry.Payload.Dispose(); if (entry.Slab is { } slab) ReleaseSlab(slab); entry.Key.Dispose(); // the reference the dictionary held, not whichever copy found it @@ -1769,10 +1771,13 @@ internal void Publish() /// (), because the copied bytes parse to the same answer. /// /// - /// The memo is not charged to : the budget meters bytes - /// the cache stores, and the managed size of an arbitrary parsed object is not knowable. For a - /// memoized string the entry's true footprint roughly doubles; a budget tuned to the knife's edge - /// should allow for that. + /// The memo is charged to when its size is known: the + /// core sizes the types it enrolls itself (a string's header-plus-UTF-16, a box), and an opted-in + /// handler volunteers an estimate through - + /// 0 meaning uncounted, because the true managed footprint of an arbitrary object graph is not + /// knowable and a wrong guess is worse than an honest shrug. The charge is paired with the + /// object itself and credited back by whoever takes it, so the budget cannot + /// drift. /// /// internal readonly struct ResultReuse @@ -1806,15 +1811,38 @@ public bool TryGet(Protocol.IRespHandler handler, out TResult /// Offer a parsed result for later hits; kept only if the slot is still empty. /// What parsing the reply produces. - public void TrySet(Protocol.IRespHandler handler, TResult value) - => (_entry as Entry)?.TrySetMemo(new Memo(handler, value)); + /// The handler whose parse this is. + /// The parsed result. + /// + /// The value's approximate retained size, charged to the byte budget; 0 leaves it uncounted. + /// + public void TrySet(Protocol.IRespHandler handler, TResult value, long bytes = 0) + { + if (_entry is not Entry entry) return; + + // the charge is paired with THIS memo object: whoever later takes the memo out of the + // entry (removal, replacement, Dispose) credits exactly memo.Bytes back, and a set that + // loses - to another setter, or to the tombstone a discard leaves - charges nothing. That + // is what keeps "credited by exactly what it was debited" true without a lock. + var memo = new Memo(handler, value, Math.Max(0, bytes)); + if (entry.TrySetMemo(memo) && memo.Bytes != 0) + { + Interlocked.Add(ref _cache!._bytes, memo.Bytes); + } + } } /// A parsed result and the handler whose parse it is; see . - private sealed class Memo(object handler, object? value) + private sealed class Memo(object handler, object? value, long bytes) { + /// The discard marker: a slot holding this can never be set again. See . + internal static readonly Memo Tombstone = new(null!, null, 0); + internal object Handler { get; } = handler; internal object? Value { get; } = value; + + /// What this memo charged to the budget; credited back by whoever takes it. + internal long Bytes { get; } = bytes; } private sealed class Entry(RespPayload payload, Dependency[] dependencies, RespRequest key, Slab? slab = null) @@ -1822,10 +1850,26 @@ private sealed class Entry(RespPayload payload, Dependency[] dependencies, RespR /// The memoized parse of this entry's payload, if anybody stored one; see . private Memo? _memo; + /// The tombstone's handler is null, which no live handler is, so readers need no special case. internal Memo? Memo => Volatile.Read(ref _memo); /// First writer wins; see for why. - internal void TrySetMemo(Memo memo) => Interlocked.CompareExchange(ref _memo, memo, null); + /// Whether this memo is the one kept - the caller charges the budget only then. + internal bool TrySetMemo(Memo memo) => Interlocked.CompareExchange(ref _memo, memo, null) is null; + + /// + /// Take the memo out, leaving the slot permanently closed. + /// + /// + /// The tombstone is what makes the budget exact under the race between a discard and a late + /// : a setter that loses to it charges nothing, and a setter that won + /// first is credited here, by the same object that carried the charge. + /// + internal Memo? TakeMemo() + { + var taken = Interlocked.Exchange(ref _memo, Memo.Tombstone); + return ReferenceEquals(taken, Memo.Tombstone) ? null : taken; + } // A single dependency - every single-key command, which is most of them - is held inline, and an array // only for more. Every hit asks IsValid, and the array was one more dependent load between the entry and @@ -1902,8 +1946,10 @@ internal Entry CopyForward(RespPayload payload, RespRequest key, Slab? slab) { var moved = new Entry(payload, _many ?? [_single], key, slab) { FilledAt = FilledAt }; - // the copied bytes parse to the same answer, so the parse survives the move - moved._memo = Volatile.Read(ref _memo); + // the copied bytes parse to the same answer, so the parse survives the move - as a + // TRANSFER of ownership, so the charge it carries is credited exactly once, by whoever + // eventually takes it from the new entry rather than by the husk this leaves behind + moved._memo = TakeMemo(); return moved; } diff --git a/src/StackExchange.Redis/Contexts/RespExecutor.cs b/src/StackExchange.Redis/Contexts/RespExecutor.cs index 393f055ab..9bab68b28 100644 --- a/src/StackExchange.Redis/Contexts/RespExecutor.cs +++ b/src/StackExchange.Redis/Contexts/RespExecutor.cs @@ -1562,7 +1562,7 @@ private static bool TryServeFromCache( if (!reuse.TryGet(handler, out result)) { result = Parse(handler, hit); - if (MayShareResult(handler, in result)) reuse.TrySet(handler, result); + if (MayShareResult(handler, in result)) reuse.TrySet(handler, result, EstimateMemoBytes(handler, in result)); } return true; @@ -1595,6 +1595,24 @@ private static bool MayShareResult(IRespHandler handler, in TR => ResultSharing.Always || (handler is Protocol.IRespResultReuse reuse && result is not null && reuse.MayReuse(in result)); + /// What a memoized result charges to the cache's byte budget; 0 is uncounted. + /// + /// The core sizes what it enrolls itself - a string is header plus UTF-16, a no-reference struct + /// is its box - and only a handler-opted result consults the handler's own + /// . Estimates, deliberately: the budget + /// wants honesty about scale, not an object-graph walker. + /// + private static long EstimateMemoBytes(IRespHandler handler, in TResult result) + { + if (typeof(TResult) == typeof(string)) + { + return result is string s ? 24 + (2L * s.Length) : 0; + } + + if (ResultSharing.Always) return ResultSharing.BoxBytes; + return handler is Protocol.IRespResultReuse reuse ? reuse.SizeOf(in result) : 0; + } + /// The per-type half of , answered once per . private static class ResultSharing { @@ -1605,6 +1623,14 @@ private static class ResultSharing #else || typeof(TResult).IsPrimitive || typeof(TResult).IsEnum; #endif + + /// What the memoized box of a no-reference struct costs: object header plus the value. + public static readonly long BoxBytes = +#if NET + 16 + System.Runtime.CompilerServices.Unsafe.SizeOf(); +#else + 24; // the fallback only enrolls primitives and enums, none larger than 8 bytes +#endif } /// diff --git a/src/StackExchange.Redis/Protocol/IRespResultReuse.cs b/src/StackExchange.Redis/Protocol/IRespResultReuse.cs index ece311451..2a5aa6013 100644 --- a/src/StackExchange.Redis/Protocol/IRespResultReuse.cs +++ b/src/StackExchange.Redis/Protocol/IRespResultReuse.cs @@ -30,5 +30,17 @@ public interface IRespResultReuse /// Whether this value may be handed to more than one caller. /// The value just parsed. bool MayReuse(in TResult value); + + /// The approximate retained size of this value, for the cache's byte budget. + /// The value just parsed; only consulted for values admitted. + /// An estimate in bytes, or 0 to leave the value uncounted. + /// + /// A memoized result lives as long as its cache entry, so an unsized one is memory the budget + /// cannot see; returning an estimate keeps + /// honest, and an estimate is all it can be - the true managed footprint of an object graph is not + /// knowable. Types the core enrolls by itself (strings, no-reference value types) are sized by the + /// core and do not consult this. + /// + long SizeOf(in TResult value); } } diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index bd1a18e6b..ca33ae50e 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -262,6 +262,7 @@ StackExchange.Redis.Protocol.IRespHandler StackExchange.Redis.Protocol.IRespHandler.Parse(ref RESPite.Messages.RespReader reader) -> TResult StackExchange.Redis.Protocol.IRespResultReuse StackExchange.Redis.Protocol.IRespResultReuse.MayReuse(in TResult value) -> bool +StackExchange.Redis.Protocol.IRespResultReuse.SizeOf(in TResult value) -> long StackExchange.Redis.Protocol.KeyRange StackExchange.Redis.Protocol.KeyRange.KeyRange() -> void StackExchange.Redis.Protocol.KeyRange.KeyRange(int offset, int length) -> void diff --git a/tests/StackExchange.Redis.Tests/RespCacheResultReuseTests.cs b/tests/StackExchange.Redis.Tests/RespCacheResultReuseTests.cs index a3a1f0b20..26054d3c3 100644 --- a/tests/StackExchange.Redis.Tests/RespCacheResultReuseTests.cs +++ b/tests/StackExchange.Redis.Tests/RespCacheResultReuseTests.cs @@ -196,11 +196,15 @@ private sealed class Poco(int length) private sealed class ReusablePocoHandler : IRespHandler, IRespResultReuse { + public const long DeclaredSize = 123; + public static readonly ReusablePocoHandler Instance = new(); public Poco Parse(ref RespReader reader) => new(reader.ScalarLength()); public bool MayReuse(in Poco value) => true; + + public long SizeOf(in Poco value) => DeclaredSize; } private sealed class PlainPocoHandler : IRespHandler @@ -225,10 +229,79 @@ public bool MayReuse(in Poco value) Interlocked.Increment(ref _asked); return false; } + + public long SizeOf(in Poco value) => 0; } private sealed class ScalingHandler(int factor) : IRespHandler { public int Parse(ref RespReader reader) => reader.ScalarLength() * factor; } + + private sealed class UnsizedPocoHandler : IRespHandler, IRespResultReuse + { + public static readonly UnsizedPocoHandler Instance = new(); + + public Poco Parse(ref RespReader reader) => new(reader.ScalarLength()); + + public bool MayReuse(in Poco value) => true; + + public long SizeOf(in Poco value) => 0; // shareable, but declines to guess a size + } + + [Fact] + public async Task AMemoizedStringIsChargedToTheBudget() + { + var (context, cache, _) = Create("$5\r\nhello\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + var afterFill = cache.Bytes; + + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); // memoizes + Assert.Equal(24 + (2 * 5), cache.Bytes - afterFill); // header + UTF-16, the core's string estimate + + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); // served; no re-charge + Assert.Equal(24 + (2 * 5), cache.Bytes - afterFill); + } + } + + [Fact] + public async Task AVolunteeredSizeIsChargedAndCreditedBack() + { + var (context, cache, _) = Create("$5\r\nhello\r\n", "+OK\r\n", "$5\r\nworld\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, ReusablePocoHandler.Instance); + var afterFill = cache.Bytes; + + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, ReusablePocoHandler.Instance); + Assert.Equal(ReusablePocoHandler.DeclaredSize, cache.Bytes - afterFill); + + // our own write discards the entry; the refill's entry must carry no stale memo charge - + // the credit is paired with the memo object, so the budget returns to one entry + one memo + _ = await context.SendAsync($"{RedisCommand.SET}{(RedisKey)"k"}{(RedisValue)"world"}", CommandFlags.None); + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, ReusablePocoHandler.Instance); + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, ReusablePocoHandler.Instance); + + Assert.Equal(afterFill + ReusablePocoHandler.DeclaredSize, cache.Bytes); + } + } + + [Fact] + public async Task AZeroSizeIsReusedButUncounted() + { + var (context, cache, _) = Create("$5\r\nhello\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, UnsizedPocoHandler.Instance); + var afterFill = cache.Bytes; + + var parsed = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, UnsizedPocoHandler.Instance); + var reused = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable, UnsizedPocoHandler.Instance); + + Assert.Same(parsed, reused); // still shared... + Assert.Equal(afterFill, cache.Bytes); // ...just invisible to the budget, as 0 asks + } + } } From 99ae40b7af93a9938c7fa1db1c039dc31a58ce30 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Thu, 12 Nov 2026 13:50:39 +0000 Subject: [PATCH 3/3] Per-key removal notification for the client-side cache (#3350) * Per-key removal notification for the client-side cache A context enrolled via RespDatabaseContext.WithRemovalHandler is told - with the key and a reason - when a key it read stops being servable from the client-side cache, so layers above (HybridCache tag memos, backplane adapters) can keep derived state coherent without polling. Observers hang on the key-table NODE, the one object every death already touches: invalidation fires promptly while the corpse is still resident, flush walks the nodes it already walks, and slab compaction is not a removal. Enrollment is one-shot - armed by a serve, consumed by the first removal at or after it, re-armed by reading again - which makes exactly-once structural (an Interlocked take, as the result memo does) and bounds dangling handlers. Arm-then-recheck closes the serve/death race, and a value handed out already dead (a fill that raced an invalidation, a stale-grace serve) is announced immediately rather than silently. Dispatch is the library's existing model: never on the caller's thread - the socket reader, the table's locks, the evictor - but queued to the thread pool like pub/sub and connection events, with exceptions swallowed per handler. No key copy: Node.Key is already immortal. Reasons separate value-changed (Invalidated, LocalWrite, Replaced) from cache hygiene (Evicted, Expired) and the bulk case (Flushed); a correctness-only consumer treats them all as drop. A fill whose tracking prefix was dropped enrolls nothing: no invalidation is coming for it to hear. * Per-caller cache behaviour is one policy, not a wither per knob WithMaxCacheAge and WithRemovalHandler are gone; the context carries a whole CachePolicy (WithCachePolicy, on both public context types), and CachePolicy - now a record, so callers derive with 'policy with {...}' - gains RemovalHandler. The surface stops growing contexts-times-knobs: a new read-time knob is one init property, and the first ship had already dropped a cell of that matrix (no WithRemovalHandler on RespServerContext). A per-read policy replaces the default wholesale and narrows, never widens: the default TimeToLive stays a ceiling, reclamation follows the default policy alone, and a per-read grace window only functions when the default policy enabled stamping. RefreshAfter is now genuinely per-caller. A RemovalHandler on the DefaultPolicy is connection-wide enrollment - every read arms it, including background refresh completions - which is what a backplane-style consumer wants, and has a new test. The policy's tick conversions are memoised at construction (the old MaxCacheAgeService did this; now the default-policy hit path gets it too), which surfaced a static-ordering trap: Default must be declared after the field its construction reads. --- docs/ClientSideCaching.md | 22 +- .../Caching/CacheKeyRemoval.cs | 65 +++ .../Caching/CachePolicy.cs | 144 ++++-- .../Caching/RespClientCache.Slabs.cs | 11 +- .../Caching/RespClientCache.cs | 215 ++++++-- .../Caching/RespKeyTable.cs | 153 +++++- .../Contexts/RespContext.cs | 28 +- .../Contexts/RespDatabaseContext.cs | 27 +- .../Contexts/RespExecutor.cs | 24 +- .../Contexts/RespServerContext.cs | 5 +- .../PublicAPI/PublicAPI.Unshipped.txt | 22 +- .../RespCacheLifetimeTests.cs | 10 +- .../RespCacheRemovalTests.cs | 465 ++++++++++++++++++ .../RespStaleWhileRevalidateTests.cs | 9 +- 14 files changed, 1071 insertions(+), 129 deletions(-) create mode 100644 src/StackExchange.Redis/Caching/CacheKeyRemoval.cs create mode 100644 tests/StackExchange.Redis.Tests/RespCacheRemovalTests.cs diff --git a/docs/ClientSideCaching.md b/docs/ClientSideCaching.md index 99ef7b2a7..a25b104de 100644 --- a/docs/ClientSideCaching.md +++ b/docs/ClientSideCaching.md @@ -122,8 +122,9 @@ Controlling it | `TimeToLive` | 1 minute | The longest an entry is ever served, whether or not it was invalidated. A safety bound against missed invalidations and key expiry. | | `RefreshAfter` | off | Once an entry is this old, a read still gets it, *and* a refresh is started in the background - so a hot key never goes from "cached" to "missing" for everyone at once. | | `InvalidationGracePeriod` | off | After *another client's* change invalidates an entry, keep serving the old value for up to this long while a refresh runs. This deliberately serves a value the server has said is out of date, to protect a very hot key from every reader missing at the same instant. It never applies to your own writes. | +| `RemovalHandler` | none | An observer told when a key read under this policy stops being servable - invalidated, replaced, evicted, expired or flushed - for layers that keep derived state coherent with cached keys. On the `DefaultPolicy` it hears about every cached read; on a per-caller policy (see [below](#per-caller-policy-withcachepolicy)) only about that caller's. | -Entry ages - `TimeToLive`, `RefreshAfter` and `WithMaxCacheAge` - are measured on a coarse clock that is cheap enough to read on every hit, so they are accurate to a few milliseconds (about 16 ms on Windows). `InvalidationGracePeriod` is measured precisely. +Everything in a `CachePolicy` is applied when an entry is *read*, never stamped when it is stored - the entry is shared, and one copy serves callers with different tolerances. Entry ages - `TimeToLive` and `RefreshAfter` - are measured on a coarse clock that is cheap enough to read on every hit, so they are accurate to a few milliseconds (about 16 ms on Windows). `InvalidationGracePeriod` is measured precisely. ### Serving stale while refreshing @@ -139,7 +140,7 @@ How the window behaves, for both: - **A failed refresh is retried by a later read.** The old value goes on being served for the rest of the window, then the entry expires as usual. - **A change wins over a refresh in flight.** If the key changes while a refresh is on the wire, the refresh's reply is not stored, exactly as for an ordinary fill. - **The grace period is measured from the invalidation**, not from the first read to notice it, and it is the cap. A key nobody reads during the window simply expires - nothing resurrects it later - and on a key written faster than it can be refreshed, stale answers stop when the window ends rather than continuing indefinitely. -- **Neither extends an entry's life.** `TimeToLive` remains the longest any entry is served, and a context's `WithMaxCacheAge` is honoured inside the window too: a caller who asked for fresher gets a miss, not a stale answer. +- **Neither extends an entry's life.** `TimeToLive` remains the longest any entry is served, and a caller's own policy (`WithCachePolicy`) is honoured inside the window too: a caller who asked for fresher gets a miss, not a stale answer. When the window does **not** apply: @@ -271,13 +272,24 @@ What that gives you: var hot = await db.Strings.GetAsync(key, CommandFlags.CacheOptIn); ``` -- **`WithMaxCacheAge(TimeSpan)`** - a context that accepts cached answers only up to a given age. It can only tighten `TimeToLive`, never extend it: +- **`WithCachePolicy(CachePolicy)`** - a context whose reads follow their own policy instead of the cache's `DefaultPolicy`. It *replaces* the default for those reads rather than merging with it, so derive it from the policy you mean (`policy with { ... }`), and it narrows rather than widens: the default policy's `TimeToLive` remains a ceiling, and how long the shared entry actually lives stays the cache's own business. A caller who wants very fresh answers, for instance: ```csharp - var fresh = db.Context.WithMaxCacheAge(TimeSpan.FromSeconds(1)); + var fresh = db.Context.WithCachePolicy(new CachePolicy { TimeToLive = TimeSpan.FromSeconds(1) }); var value = await fresh.Strings.GetAsync(key); ``` + The same policy object carries a `RemovalHandler`, for a caller that wants to hear when the keys *it* read stop being current: + + ```csharp + var observed = db.Context.WithCachePolicy(new CachePolicy + { + RemovalHandler = static (key, reason) => InvalidateDerivedState(key), + }); + ``` + + Enrollment rides the read: each read arms one notification for that key, the first removal at or after it fires it (from the thread pool, never the connection's reader), and reading again re-arms. Contexts and policies are cheap and immutable - build them once and reuse them. + - **`WithoutCache()`** - a context that bypasses the cache entirely: ```csharp @@ -314,5 +326,5 @@ See also - [Using RESP3](Resp3) - [Configuration](Configuration) -- [Command groups](Groups) - contexts, and where `WithMaxCacheAge` / `WithoutCache` fit +- [Command groups](Groups) - contexts, and where `WithCachePolicy` / `WithoutCache` fit - [Redis: client-side caching](https://redis.io/docs/latest/develop/reference/client-side-caching/) diff --git a/src/StackExchange.Redis/Caching/CacheKeyRemoval.cs b/src/StackExchange.Redis/Caching/CacheKeyRemoval.cs new file mode 100644 index 000000000..54de2ffee --- /dev/null +++ b/src/StackExchange.Redis/Caching/CacheKeyRemoval.cs @@ -0,0 +1,65 @@ +using System; + +namespace StackExchange.Redis.Caching +{ + /// + /// Why a key the caller had read through the client-side cache stopped being servable. + /// + /// + /// + /// A correctness-only consumer treats every reason as "drop". The split exists for smarter + /// consumers, separating "the value changed, or may have" (, + /// , ) from cache hygiene the server knows nothing + /// about (, ) and from the bulk case + /// (). + /// + /// + public enum CacheKeyRemovalReason + { + /// The server said the key changed: an invalidation push or keyspace notification. + Invalidated, + + /// This process wrote the key itself - a stronger statement than a server push, because it is a fact rather than a race. + LocalWrite, + + /// The cached value was replaced by a fresh fill; the value that was read is gone. + Replaced, + + /// The entry aged past the cache's lifetime - the backstop against missed invalidations - and was reclaimed. + /// + /// Expiry is decided lazily, so this fires at physical reclamation rather than at the moment the + /// lifetime passed. A consumer with its own freshness bound should keep enforcing it; this reason + /// is the backstop's backstop, not a timer. + /// + Expired, + + /// The entry was evicted to stay inside the cache's budget; the server may still be tracking the key. + Evicted, + + /// Everything went at once: a server flush, a lost connection, or tracking being torn down. + Flushed, + } + + /// + /// Told when a key read under an enrolled policy stops being servable from the client-side + /// cache; see . + /// + /// + /// The key exactly as stored - prefixes included - valid only for the duration of the callback. + /// + /// Why the key stopped being servable. + /// + /// + /// Called on a thread-pool thread, never on the connection's reader - the same dispatch model + /// as pub/sub handlers and connection events - so being notified cannot stall the connection. + /// Exceptions are swallowed: a notification is advice, and advice must not break the cache. + /// + /// + /// One enrollment, one notification. An enrollment is armed by a serve and consumed by the + /// first removal at or after it; reading the key again re-arms it. A notification can therefore + /// arrive after the consumer has already re-read the key - dropping derived state is always safe, so + /// a late notification costs a rebuild, never correctness. + /// + /// + public delegate void CacheKeyRemovedHandler(ReadOnlySpan key, CacheKeyRemovalReason reason); +} diff --git a/src/StackExchange.Redis/Caching/CachePolicy.cs b/src/StackExchange.Redis/Caching/CachePolicy.cs index f34867356..a96e88e8f 100644 --- a/src/StackExchange.Redis/Caching/CachePolicy.cs +++ b/src/StackExchange.Redis/Caching/CachePolicy.cs @@ -18,19 +18,33 @@ namespace StackExchange.Redis.Caching /// about the connection and live on instead. /// /// - /// The freshness tolerance a particular caller will accept rides on the context - see - /// - and is read-time for exactly the same reason. + /// The same read-time nature is what lets a whole policy ride on the context - see + /// - so a caller with a different freshness + /// tolerance, or a removal observer to arm, states it per call without changing how the shared entry + /// behaves for anyone else. A per-read policy replaces the cache's + /// for those reads rather than merging with it, so derive it + /// from the policy you mean: policy with { TimeToLive = ... }. /// /// - /// It also keeps at its 48 bytes: a context carries a reference to shared - /// policy plus at most one override, rather than a field per knob. + /// It also keeps at its 48 bytes: a context carries a reference to a shared + /// policy, rather than a field per knob. /// /// - public sealed class CachePolicy + public sealed record CachePolicy { + // BEFORE Default: statics initialise in declaration order, and Default's construction reads this + private static readonly TimeSpan DefaultTimeToLive = TimeSpan.FromMinutes(1); + /// The default policy, used when none is given. public static CachePolicy Default { get; } = new(); + private readonly TimeSpan _timeToLive = DefaultTimeToLive; + private readonly TimeSpan _refreshAfter; + private readonly TimeSpan _invalidationGracePeriod; + private readonly long _timeToLiveTicks = ToTicks(DefaultTimeToLive); + private readonly long _refreshAfterTicks; + private readonly long _serveStaleTicks; + /// /// The longest an entry may be served after it was fetched, regardless of invalidation. /// @@ -49,8 +63,27 @@ public sealed class CachePolicy /// seconds, so the default of one minute is comfortably longer than detection while still bounding /// the damage when detection itself fails. /// + /// + /// On a per-read policy () this is the caller's + /// own freshness requirement, and it narrows, never widens: the + /// 's lifetime remains a ceiling, and reclamation follows + /// the default policy alone - a read policy changes what its own reads will accept, never how long + /// the shared entry lives. + /// /// - public TimeSpan TimeToLive { get; init; } = TimeSpan.FromMinutes(1); + public TimeSpan TimeToLive + { + get + { + return _timeToLive; + } + + init + { + _timeToLive = value; + _timeToLiveTicks = ToTicks(value); + } + } /// /// How old an entry may get before a read refreshes it in the background, while still being served. @@ -75,7 +108,19 @@ public sealed class CachePolicy /// retained to make it possible. /// /// - public TimeSpan RefreshAfter { get; init; } = TimeSpan.Zero; + public TimeSpan RefreshAfter + { + get + { + return _refreshAfter; + } + + init + { + _refreshAfter = value; + _refreshAfterTicks = ToTicks(value); + } + } /// /// A grace period after an invalidation, during which the old value may still be served while a @@ -117,8 +162,61 @@ public sealed class CachePolicy /// which is otherwise a few nanoseconds wide and sees every key the server mentions - so the cost /// lands only on those who asked for the feature. /// + /// + /// That stamping decision belongs to the , because the + /// invalidation path cannot know which context will read next. So on a per-read policy this window + /// only functions when the default policy enabled one too; given that, the window used is this + /// policy's own - a caller can opt out of stale serving, or use a different window, still inside + /// . + /// + /// + public TimeSpan InvalidationGracePeriod + { + get + { + return _invalidationGracePeriod; + } + + init + { + _invalidationGracePeriod = value; + _serveStaleTicks = ToPreciseTicks(value); + } + } + + /// + /// A removal observer: reads that follow this policy arm it on the keys they serve, and it is told + /// when such a key stops being servable from the client-side cache. + /// + /// + /// + /// You hear about the keys you read, nothing else. Enrollment rides the read: any serve + /// under this policy - a cache hit, or the fill a miss performs - arms the observer on the keys it + /// depended on, and the first removal at or after that consumes the enrollment. Reading again + /// re-arms it. A key may have entered the cache through a context with no observer at all: a later + /// hit under this policy still enrolls. + /// + /// + /// Per enrollment, exactly one notification, and never a missed removal - including a fill + /// that raced an invalidation, and a value served from the stale-grace window, both of which fire + /// immediately: a value that is already not current is announced, not silently handed over. A + /// notification can arrive after a racing re-read, so treat it as "drop derived state", which is + /// always safe; it is advice about staleness, not an ordered event stream. + /// + /// + /// The observer is told the key (exactly as stored, prefixes included, valid only for the + /// callback) and why it went. It is called from the thread pool, never from the connection's + /// reader. Keep it long-lived and cheap: build the policy once and reuse it (the handler + /// reference is the identity enrollment dedupes on), and do only bookkeeping in the callback - it + /// shares the thread pool with everything else. Exceptions are swallowed. + /// + /// + /// On the it is connection-wide: every cached read + /// enrolls it, and so does a background refresh completing - so a watched key stays watched across + /// its own replacement, for as long as anything keeps reading it. + /// /// - public TimeSpan InvalidationGracePeriod { get; init; } = TimeSpan.Zero; + public CacheKeyRemovedHandler? RemovalHandler { get; init; } /// Whether this policy asks for background refresh at all. internal bool RefreshesEarly => RefreshAfter > TimeSpan.Zero && RefreshAfter < TimeToLive; @@ -131,13 +229,13 @@ public sealed class CachePolicy /// the ages below: a grace window can be tens of milliseconds, and it is only consulted for an entry that /// has already been invalidated, so its clock read is off the hit path. /// - internal long ServeStaleTicks => ToPreciseTicks(InvalidationGracePeriod); + internal long ServeStaleTicks => _serveStaleTicks; - /// in ticks. - internal long RefreshAfterTicks => ToTicks(RefreshAfter); + /// in ticks; memoised, the hit path reads it per serve. + internal long RefreshAfterTicks => _refreshAfterTicks; - /// in ticks. - internal long TimeToLiveTicks => ToTicks(TimeToLive); + /// in ticks; memoised, the hit path reads it per serve. + internal long TimeToLiveTicks => _timeToLiveTicks; /// A span in ticks: the coarse clock that entry ages are measured on. internal static long ToTicks(TimeSpan value) @@ -189,24 +287,4 @@ internal static class CacheClock internal static long Now => Stopwatch.GetTimestamp(); #endif } - - /// - /// A per-context override of how stale an answer the caller will accept. - /// - /// - /// The one piece of cache configuration that cannot be added to the existing surface: - /// IDatabase.StringGet cannot grow a parameter without a binary break, whereas the context - /// reaches every command without touching a single signature. - /// - /// Applied when the entry is read, never stamped when it is stored - the entry is shared, so one - /// copy has to serve callers with different tolerances. Stamping at store time would force identical - /// replies to be cached once per distinct lifetime, which is the opposite of what a shared cache is for. - /// - /// - internal sealed class MaxCacheAgeService(TimeSpan maxAge) - { - internal TimeSpan MaxAge { get; } = maxAge; - - internal long Ticks { get; } = CachePolicy.ToTicks(maxAge); - } } diff --git a/src/StackExchange.Redis/Caching/RespClientCache.Slabs.cs b/src/StackExchange.Redis/Caching/RespClientCache.Slabs.cs index 3b6caf9a2..3f6f938ac 100644 --- a/src/StackExchange.Redis/Caching/RespClientCache.Slabs.cs +++ b/src/StackExchange.Redis/Caching/RespClientCache.Slabs.cs @@ -268,7 +268,12 @@ private bool TryEvictOldestSlab(ref int evicted) } else if (entries.Remove(new KeyValuePair(key, entry))) { - Release(entry); + // aged out, or live-but-unread in the slab being taken: Expired or Evicted says which; + // an invalidated one already announced itself and Release says nothing for it + var reason = CachePolicy.IsOlderThan(entry.FilledAt, lifetime) + ? CacheKeyRemovalReason.Expired + : CacheKeyRemovalReason.Evicted; + Release(entry, reason); evicted++; } } @@ -320,8 +325,8 @@ private void Promote(in EntryKey key, Entry entry, Slab from, ref int evicted) } // a fill re-added the key between the remove and the add; theirs is as good as ours, and the old entry - // is simply gone - Release(entry); + // is simply gone - superseded by that fill, which is what Replaced means + Release(entry, CacheKeyRemovalReason.Replaced); if (slab is not null) ReleaseSlab(slab); evicted++; } diff --git a/src/StackExchange.Redis/Caching/RespClientCache.cs b/src/StackExchange.Redis/Caching/RespClientCache.cs index b4f63583a..29c863aff 100644 --- a/src/StackExchange.Redis/Caching/RespClientCache.cs +++ b/src/StackExchange.Redis/Caching/RespClientCache.cs @@ -475,7 +475,7 @@ private static bool IsReadOnly(CommandFlags flags) /// Fills refused because the reply was an error. /// - /// See for why errors are not cacheable. A non-trivial count here is + /// See for why errors are not cacheable. A non-trivial count here is /// worth investigating on its own: errors should be rare, and caching them would have hidden that. /// public long RefusedError => Volatile.Read(ref _refusedError); @@ -646,20 +646,20 @@ internal void OnLocalFlush() /// the parse is done. /// public bool TryGet(in RespRequest frame, int database, [NotNullWhen(true)] out RespPayload? payload) - => TryGet(in frame, database, long.MaxValue, out payload); + => TryGet(in frame, database, policy: null, out payload); - /// Look for a cached response, subject to a freshness requirement. + /// Look for a cached response, subject to a caller's own read policy. /// The rendered request. /// The database the request ran against. - /// - /// The caller's own freshness requirement, from ; - /// when they did not state one. + /// + /// The caller's own read policy, from ; null when they + /// did not state one, meaning the cache's . /// /// The cached reply, retained. /// /// - /// Age is checked on read, against the stricter of the policy's lifetime and the caller's - /// requirement - never stamped when the entry was stored. The entry is shared, so one copy serves + /// Age is checked on read, against the stricter of the cache policy's lifetime and the + /// caller's - never stamped when the entry was stored. The entry is shared, so one copy serves /// callers with different tolerances; stamping at store time would force identical replies to be /// cached once per distinct lifetime, which is the opposite of what a shared cache is for. /// @@ -668,13 +668,13 @@ public bool TryGet(in RespRequest frame, int database, [NotNullWhen(true)] out R /// a read path, and is where entries go. /// /// - public bool TryGet(in RespRequest frame, int database, long maxAgeTicks, [NotNullWhen(true)] out RespPayload? payload) - => TryGet(in frame, database, maxAgeTicks, out payload, out _); + public bool TryGet(in RespRequest frame, int database, CachePolicy? policy, [NotNullWhen(true)] out RespPayload? payload) + => TryGet(in frame, database, policy, out payload, out _); - /// + /// /// The rendered request. /// The database the request ran against. - /// The caller's own freshness requirement. + /// The caller's own read policy, or null for the cache's. /// The cached reply, retained. /// /// true if this caller has claimed the job of refreshing an ageing entry, and must @@ -688,15 +688,19 @@ public bool TryGet(in RespRequest frame, int database, long maxAgeTicks, [NotNul public bool TryGet( in RespRequest frame, int database, - long maxAgeTicks, + CachePolicy? policy, [NotNullWhen(true)] out RespPayload? payload, out bool shouldRefresh) - => TryGet(in frame, database, maxAgeTicks, out payload, out shouldRefresh, out _); + => TryGet(in frame, database, policy, out payload, out shouldRefresh, out _); - /// + /// /// The rendered request. /// The database the request ran against. - /// The caller's own freshness requirement. + /// + /// The caller's own read policy, or null for the cache's. Its + /// , when it has one, is armed only when something is + /// actually served - a miss arms nothing, because the caller holds nothing that could go stale. + /// /// The cached reply, retained. /// Whether this caller claimed the refresh; see the other overload. /// @@ -706,13 +710,15 @@ public bool TryGet( public bool TryGet( in RespRequest frame, int database, - long maxAgeTicks, + CachePolicy? policy, [NotNullWhen(true)] out RespPayload? payload, out bool shouldRefresh, out ResultReuse reuse) { shouldRefresh = false; reuse = default; + policy ??= Policy; + var onRemoved = policy.RemovalHandler; if (!_entries.TryGetValue(new EntryKey(frame, database), out var entry)) { @@ -724,9 +730,19 @@ public bool TryGet( // decision rather than a shortcut if (!entry.IsValid) { - if (TryServeStale(entry, maxAgeTicks, out payload, out shouldRefresh)) + if (TryServeStale(entry, policy, out payload, out shouldRefresh)) { reuse = new ResultReuse(this, entry); + if (onRemoved is not null) + { + // the caller is being handed a value that is ALREADY not current, so the honest + // thing an observer can be told is exactly that, now - waiting for a removal that + // has already happened would wait for ever. Arm-then-fire keeps the one-shot + // contract: this enrollment is consumed immediately, and the next read re-arms. + entry.Enroll(onRemoved); + entry.NotifyRemoved(CacheKeyRemovalReason.Invalidated); + } + return true; } @@ -739,14 +755,15 @@ public bool TryGet( // let one stale read through the door it had already closed if (entry.IsValid) { - var limit = Math.Min(Policy.TimeToLiveTicks, maxAgeTicks); + // the stricter of the cache's lifetime and the caller's: a read policy narrows, never widens + var limit = Math.Min(Policy.TimeToLiveTicks, policy.TimeToLiveTicks); if (!CachePolicy.IsOlderThan(entry.FilledAt, limit)) { // served either way; the only question is whether this reader also goes and gets a - // newer one. Note the soft threshold is the POLICY's, not the caller's: a caller - // asking for fresher than it gets a miss, which is a stronger answer than a refresh. - if (Policy.RefreshesEarly - && CachePolicy.IsOlderThan(entry.FilledAt, Policy.RefreshAfterTicks) + // newer one. The soft threshold is the read policy's own: past the caller's own + // LIFETIME they get a miss, which is a stronger answer than a refresh. + if (policy.RefreshesEarly + && CachePolicy.IsOlderThan(entry.FilledAt, policy.RefreshAfterTicks) && entry.TryClaimRefresh()) { Interlocked.Increment(ref _refreshes); @@ -758,6 +775,19 @@ public bool TryGet( entry.Touch(); payload = entry.Payload; reuse = new ResultReuse(this, entry); + + if (onRemoved is not null) + { + // ARM, then look again: an invalidation landing between the recheck above and + // the arm would have taken a set this observer was not yet in, and the caller + // would hold a value whose death was never announced. Arm first and the race + // inverts: either the invalidation's take sees this observer, or this second + // look sees the invalidation and fires it here. The serve itself stands - it + // passed the same recheck every un-enrolled caller relies on. + entry.Enroll(onRemoved); + if (!entry.IsValid) entry.NotifyRemoved(CacheKeyRemovalReason.Invalidated); + } + return true; } @@ -799,23 +829,27 @@ public bool TryGet( /// /// /// Within the same age limits as a valid entry: , the backstop - /// against missed invalidations, and the caller's own . The grace + /// against missed invalidations, from both the cache's policy and the caller's own. The grace /// period shortens an entry's life after an invalidation; it never extends it. /// /// + /// The policy consulted is the CALLER's (): a caller can opt out of stale + /// serving, or use its own window - but only when the cache's own enabled the + /// feature too, because that is what turns on the invalidation-time stamping this is measured + /// against; unstamped, is zero and every window refuses. /// - private bool TryServeStale(Entry entry, long maxAgeTicks, [NotNullWhen(true)] out RespPayload? payload, out bool shouldRefresh) + private bool TryServeStale(Entry entry, CachePolicy policy, [NotNullWhen(true)] out RespPayload? payload, out bool shouldRefresh) { shouldRefresh = false; payload = null; - if (!Policy.ServesStale || entry.WrittenLocally) return false; + if (!policy.ServesStale || entry.WrittenLocally) return false; // measured from the INVALIDATION, not from whenever somebody first looked: a key nobody has read // for an hour should expire, not be resurrected by the next reader to wander past var staleSince = entry.StaleSince; - if (staleSince == 0 || CachePolicy.IsOlderThanPrecise(staleSince, Policy.ServeStaleTicks)) return false; - if (CachePolicy.IsOlderThan(entry.FilledAt, Math.Min(Policy.TimeToLiveTicks, maxAgeTicks))) return false; + if (staleSince == 0 || CachePolicy.IsOlderThanPrecise(staleSince, policy.ServeStaleTicks)) return false; + if (CachePolicy.IsOlderThan(entry.FilledAt, Math.Min(Policy.TimeToLiveTicks, policy.TimeToLiveTicks))) return false; if (!entry.Payload.TryRetain()) return false; @@ -1128,15 +1162,44 @@ internal void Unregister(in RespFill fill) /// copied. It is already in a pooled, reference-counted buffer; copying it into a second one to /// cache it would be pure waste. /// - public bool TryComplete(in RespFill fill, RespPayload response) + public bool TryComplete(in RespFill fill, RespPayload response) => TryComplete(in fill, response, policy: null); + + /// + /// The fill to complete. + /// The reply that arrived. + /// + /// The read policy of the caller this fill serves, or null for the cache's own. Its + /// , when it has one, is armed on the filled keys whether + /// or not the reply was stored - the server is tracking these keys either way - EXCEPT when it + /// provably is not (the tracking prefix was dropped) or the reply was an error, where "you will + /// be told" would be a promise nothing can keep. + /// + public bool TryComplete(in RespFill fill, RespPayload response, CachePolicy? policy) { if (response is null) throw new ArgumentNullException(nameof(response)); if (fill.Key.IsEmpty) return false; + var onRemoved = (policy ?? Policy).RemovalHandler; Unregister(in fill); try { - return TryCompleteCore(in fill, response); + var stored = TryCompleteCore(in fill, response, out var tracked); + if (tracked && onRemoved is not null) + { + // LAST, after any store - and in particular after the store's own notification of the + // entry it replaced, or this fill's consumer would be told its brand-new value is gone. + // Arm, then look: a key that died in flight (or during the store) fires here and now, + // because its removal already happened and will not announce itself again. That covers + // the fill that raced an invalidation, which is otherwise the easiest death to miss. + var dependencies = fill.Dependencies; + foreach (var dependency in dependencies) dependency.Enroll(onRemoved); + foreach (var dependency in dependencies) + { + if (!dependency.IsValid) dependency.NotifyRemoved(CacheKeyRemovalReason.Invalidated); + } + } + + return stored; } finally { @@ -1147,12 +1210,15 @@ public bool TryComplete(in RespFill fill, RespPayload response) } } - private bool TryCompleteCore(in RespFill fill, RespPayload response) + private bool TryCompleteCore(in RespFill fill, RespPayload response, out bool tracked) { + tracked = true; if (response.PrefixDropped) { // the server never saw the CLIENT CACHING YES, so under an opt-in mode it is not tracking these keys; - // stored, this reply would sit here until eviction with no invalidation ever coming + // stored, this reply would sit here until eviction with no invalidation ever coming - and for the + // same reason, no removal observer is armed: no invalidation is coming for it to hear + tracked = false; Interlocked.Increment(ref _refusedUntracked); fill.Key.Dispose(); return false; @@ -1160,6 +1226,9 @@ private bool TryCompleteCore(in RespFill fill, RespPayload response) if (!IsCacheableReply(response.Span)) { + // no observer either: whether the server tracks keys named by an erroring command is not + // specified, and an enrollment that might never fire is worse than none + tracked = false; Interlocked.Increment(ref _refusedError); fill.Key.Dispose(); return false; @@ -1214,6 +1283,17 @@ private bool TryCompleteCore(in RespFill fill, RespPayload response) && (fill.Replaces || !previous.IsValid || CachePolicy.IsOlderThan(previous.FilledAt, Policy.TimeToLiveTicks)) && _entries.TryUpdate(entryKey, entry, previous)) { + // observers armed on the OLD value's serves: what they read is gone. Only when the previous + // entry was still valid - an invalidated one already announced itself, and firing again here + // would take enrollments that belong to the key's NEXT life (a fill arms after this returns). + // Expired-but-valid is the missed-invalidation backstop doing its job, and says so. + if (previous.IsValid) + { + previous.NotifyRemoved(CachePolicy.IsOlderThan(previous.FilledAt, Policy.TimeToLiveTicks) + ? CacheKeyRemovalReason.Expired + : CacheKeyRemovalReason.Replaced); + } + Interlocked.Add(ref _bytes, entry.Bytes - previous.Bytes); if (previous.TakeMemo() is { Bytes: not 0 } memo) Interlocked.Add(ref _bytes, -memo.Bytes); previous.Payload.Dispose(); // the superseded reply @@ -1271,7 +1351,9 @@ public int Sweep() if (entry.IsValid && !CachePolicy.IsOlderThan(entry.FilledAt, lifetime)) continue; if (_entries.TryRemove(pair.Key, out var removing)) { - Release(removing); + // an entry the sweep reclaims is a corpse: invalidated (already announced, so the + // reason goes unheard) or aged out, which is what Expired means + Release(removing, CacheKeyRemovalReason.Expired); removed++; } } @@ -1320,9 +1402,20 @@ public int SweepIfDue() /// Release() is a bare decrement with no idempotence guard. /// /// Account for an entry that has just left the dictionary, and let its payload go. - /// Called once per successful removal, which is what keeps exact. - private void Release(Entry entry) + /// + /// Called once per successful removal, which is what keeps exact. + /// + /// The reaches removal observers only when the entry was still VALID - + /// an invalidated entry announced itself when the invalidation landed, and anything armed since + /// belongs to the key's next life. Null skips the notification entirely: + /// passes it because its says + /// for everything, once. + /// + /// + private void Release(Entry entry, CacheKeyRemovalReason? reason) { + if (reason is { } why && entry.IsValid) entry.NotifyRemoved(why); + Interlocked.Decrement(ref _count); Interlocked.Add(ref _bytes, -entry.Bytes); if (entry.TakeMemo() is { Bytes: not 0 } memo) Interlocked.Add(ref _bytes, -memo.Bytes); @@ -1452,7 +1545,7 @@ private bool TryEvictOne() if (!entry.IsValid || CachePolicy.IsOlderThan(entry.FilledAt, lifetime)) { // dead or expired, and we are holding its key: take it now - if (TryRemoveSampled(candidate)) return true; + if (TryRemoveSampled(candidate, CacheKeyRemovalReason.Expired)) return true; continue; } @@ -1475,18 +1568,19 @@ private bool TryEvictOne() } // live, and not read since it was stored or last examined - if (TryRemoveSampled(candidate)) return true; + if (TryRemoveSampled(candidate, CacheKeyRemovalReason.Evicted)) return true; // somebody else took it between the look and the removal; keep looking } // every live entry in the window had been read: the longest-resident of them goes, its flag now // cleared. Its re-enqueued key is left behind as a stale one, which the allowance above absorbs - return haveFirstSpared && TryRemoveSampled(firstSparedKey); + return haveFirstSpared && TryRemoveSampled(firstSparedKey, CacheKeyRemovalReason.Evicted); } /// Remove an entry named by a key sampled from the live enumeration. /// A key copied out of the live enumeration. + /// What removal observers are told, if the entry was still valid. /// Whether this call was the one that removed it. /// /// Safe to probe with despite being a copy taken from a moving enumeration, because a cache key @@ -1495,11 +1589,11 @@ private bool TryEvictOne() /// ObjectDisposedException from the evictor, and the same race on the read path resolved as a /// lookup comparing against bytes that were back in the pool. /// - private bool TryRemoveSampled(in EntryKey sampled) + private bool TryRemoveSampled(in EntryKey sampled, CacheKeyRemovalReason reason) { if (!_entries.TryRemove(sampled, out var removed)) return false; - Release(removed); + Release(removed, reason); return true; } @@ -1508,8 +1602,9 @@ public void Dispose() { foreach (var pair in _entries) { + // no per-entry notification: the InvalidateAll below says Flushed for every armed key, once if (!_entries.TryRemove(pair.Key, out var entry)) continue; - Release(entry); + Release(entry, reason: null); } ReleaseCurrentSlab(); @@ -1684,6 +1779,12 @@ internal static bool AnyLocalWrite(Dependency[] dependencies) /// The earliest recorded invalidation of this one key, or zero; see the array form. internal long EarliestInvalidation() => IsValid ? 0 : _node.InvalidatedAt; + /// Arm a removal observer on this key's node; see . + internal void Enroll(CacheKeyRemovedHandler handler) => _node.Enroll(handler); + + /// Fire this key's armed observers; see . + internal void NotifyRemoved(CacheKeyRemovalReason reason) => _node.NotifyRemoved(reason); + internal static bool AllValid(Dependency[] dependencies) { foreach (var dependency in dependencies) @@ -1877,6 +1978,38 @@ private sealed class Entry(RespPayload payload, Dependency[] dependencies, RespR private readonly Dependency _single = dependencies.Length == 1 ? dependencies[0] : default; private readonly Dependency[]? _many = dependencies.Length == 1 ? null : dependencies; + /// Arm a removal observer on every key this entry depends on. + /// + /// On the NODES, not on this entry, so slab compaction () is not a + /// removal and invalidation fires while the corpse is still resident. A multi-key entry arms + /// each dependency independently: the notification is per KEY, so a consumer may hear once per + /// dependency that dies, each fire naming its own key. + /// + internal void Enroll(CacheKeyRemovedHandler handler) + { + if (_many is { } many) + { + foreach (var dependency in many) dependency.Enroll(handler); + } + else + { + _single.Enroll(handler); + } + } + + /// Fire the armed observers of every key this entry depends on. + internal void NotifyRemoved(CacheKeyRemovalReason reason) + { + if (_many is { } many) + { + foreach (var dependency in many) dependency.NotifyRemoved(reason); + } + else + { + _single.NotifyRemoved(reason); + } + } + /// The slab the reply lives in, or when it has an array of its own. internal Slab? Slab { get; } = slab; diff --git a/src/StackExchange.Redis/Caching/RespKeyTable.cs b/src/StackExchange.Redis/Caching/RespKeyTable.cs index cb23c1de0..f33ee97a7 100644 --- a/src/StackExchange.Redis/Caching/RespKeyTable.cs +++ b/src/StackExchange.Redis/Caching/RespKeyTable.cs @@ -66,6 +66,18 @@ internal sealed class Node private long _localWriteAt; private long _invalidatedAt; + /// + /// Removal observers armed on this key: null, one , or an + /// array of them. See and . + /// + /// + /// On the NODE rather than on any cache entry, because the node is what an invalidation + /// touches: entry release is lazy (a corpse sits until a sweep), and an observer that waited + /// for it would hold stale derived state for the whole invalid-but-resident window. The cost + /// is one field on every tracked key, paid whether or not anybody enrolls. + /// + private object? _observers; + internal Node(byte[] key, int hash, long generation) { Key = key; @@ -98,10 +110,105 @@ internal Node(byte[] key, int hash, long generation) /// rely on the timestamp already being there. /// /// - internal void Invalidate(bool stampTime) + internal void Invalidate(bool stampTime) => Invalidate(stampTime, CacheKeyRemovalReason.Invalidated); + + /// + /// Whether to record when this happened, for a grace period. + /// What armed observers are told; see . + internal void Invalidate(bool stampTime, CacheKeyRemovalReason reason) { if (stampTime) Volatile.Write(ref _invalidatedAt, Stopwatch.GetTimestamp()); Volatile.Write(ref _generation, Invalid); + + // AFTER the generation, so a handler that reacts by re-reading cannot be served the value + // it was just told about. Repeat invalidations of a dead key fire nothing: the first one + // took the observers. + NotifyRemoved(reason); + } + + /// + /// Arm a removal observer: the next removal of this key - and only the next - will tell it. + /// + /// + /// + /// One-shot, re-armed by serving. An enrollment is consumed by the first removal at or + /// after it ( takes the whole set), and the consumer's next read + /// re-arms it. That makes "exactly once per enrollment" structural - no death sequence, no + /// per-observer bookkeeping - and it bounds dangling handlers: a consumer that stops reading a + /// key is told about at most one more removal, instead of every removal for the life of the + /// connection. It also matches the consumer's own state machine: after the notification their + /// derived state is gone, and they are indifferent to further removals until they read again. + /// + /// + /// Idempotent per handler instance (reference equality), so a hot key's every hit costs a read + /// and a compare, not an allocation. Inline-one-plus-overflow, like Entry's + /// dependencies: the common case is one observer, and two consumers reading the same key + /// through different contexts must both hear. + /// + /// + internal void Enroll(CacheKeyRemovedHandler handler) + { + while (true) + { + var current = Volatile.Read(ref _observers); + object updated; + switch (current) + { + case null: + updated = handler; + break; + case CacheKeyRemovedHandler single: + if (ReferenceEquals(single, handler)) return; + updated = new[] { single, handler }; + break; + case CacheKeyRemovedHandler[] many: + foreach (var existing in many) + { + if (ReferenceEquals(existing, handler)) return; + } + + var grown = new CacheKeyRemovedHandler[many.Length + 1]; + Array.Copy(many, grown, many.Length); + grown[many.Length] = handler; + updated = grown; + break; + default: + return; // unreachable; fail safe rather than spin + } + + if (ReferenceEquals(Interlocked.CompareExchange(ref _observers, updated, current), current)) return; + } + } + + /// + /// Take every armed observer and tell them, from the thread pool. + /// + /// + /// + /// Never on the calling thread. Removals happen on the socket reader (an invalidation + /// push), under the table's own locks (), and on whoever's thread + /// evicts or sweeps - all places user code must not borrow. The same dispatch model as pub/sub + /// handlers and connection events (ConnectionMultiplexer.CompleteAsWorker): queue to + /// the pool and move on. The queueing itself is non-blocking, so firing under a lock is safe. + /// + /// + /// No key copy either: is already an immortal array - nodes are carried by + /// reference across table growth precisely because entries point at them - so the deferred + /// callback reads a span over bytes that cannot go away. + /// + /// + /// The take is the dedupe: whichever removal gets here first owns the set, and every other + /// path - a repeat invalidation, the sweep reclaiming an already-announced corpse - takes + /// null and fires nothing. + /// + /// + internal void NotifyRemoved(CacheKeyRemovalReason reason) + { + if (Volatile.Read(ref _observers) is not null + && Interlocked.Exchange(ref _observers, null) is { } taken) + { + ThreadPool.QueueUserWorkItem(PendingRemoval.Fire, new PendingRemoval(taken, Key, reason)); + } } /// When this key was last invalidated, if anybody asked for that to be recorded. @@ -138,7 +245,7 @@ internal void InvalidateLocal(bool stampTime) if (Interlocked.CompareExchange(ref _localWriteAt, ticket, current) == current) break; } - Invalidate(stampTime); + Invalidate(stampTime, CacheKeyRemovalReason.LocalWrite); } /// @@ -163,6 +270,43 @@ internal long EnsureLive() } } + /// One queued removal notification: the observers taken, the key, and why. + /// + /// Exceptions are swallowed per handler: a notification is advice, and one consumer's throw must + /// not silence another's - nor surface on a pool thread as a process-level event. + /// + private sealed class PendingRemoval(object observers, byte[] key, CacheKeyRemovalReason reason) + { + internal static readonly WaitCallback Fire = static state => ((PendingRemoval)state!).Invoke(); + + private void Invoke() + { + if (observers is CacheKeyRemovedHandler single) + { + try + { + single(key, reason); + } + catch + { + } + } + else if (observers is CacheKeyRemovedHandler[] many) + { + foreach (var handler in many) + { + try + { + handler(key, reason); + } + catch + { + } + } + } + } + } + [MethodImpl(MethodImplOptions.AggressiveInlining)] private static int HashOf(ReadOnlySpan key) => RedisValue.GetHashCode(key); @@ -224,7 +368,10 @@ internal void InvalidateAll() foreach (var bucket in _buckets) { if (bucket is null) continue; - foreach (var node in bucket) node.Invalidate(); // stamp before dropping; see the type remarks + + // stamp before dropping; see the type remarks. Observers hear Flushed promptly - the + // dispatch only QUEUES under these locks, no user code runs here + foreach (var node in bucket) node.Invalidate(stampTime: false, CacheKeyRemovalReason.Flushed); } _buckets = new Node[]?[_buckets.Length]; diff --git a/src/StackExchange.Redis/Contexts/RespContext.cs b/src/StackExchange.Redis/Contexts/RespContext.cs index 635c5d171..9109ac315 100644 --- a/src/StackExchange.Redis/Contexts/RespContext.cs +++ b/src/StackExchange.Redis/Contexts/RespContext.cs @@ -67,7 +67,7 @@ internal RespContext( _cache = TryGetService(out var cache) ? cache : null; ScriptCache = TryGetService(out var scripts) ? scripts : null; ChannelPrefix = TryGetService(out var prefix) ? prefix.Channel : default; - MaxCacheAgeTicks = TryGetService(out var maxAge) ? maxAge.Ticks : long.MaxValue; + CachePolicy = TryGetService(out var policy) ? policy : null; IsBlocking = TryGetService(out _); } @@ -676,23 +676,17 @@ private RespRequestFrame Render(string command, ReadOnlySpan ar } } - /// A context whose cached answers must be no older than . - /// The oldest answer this caller will accept. - /// - /// Narrows, never widens: the cache's own is a ceiling, and - /// this cannot raise it. So a caller can ask for fresher, never for staler than the deployment - /// allows. - /// - /// This is the one knob that belongs on the context rather than on the policy, because freshness - /// tolerance is a property of the call and not of the connection or the deployment - and it is the - /// one that could not be added to IDatabase at all without a binary break. - /// - /// - internal RespContext WithMaxCacheAge(TimeSpan maxAge) - => WithServices(new MaxCacheAgeService(maxAge)); + /// + /// A copy of this context whose cached reads follow instead of the + /// cache's ; see + /// for the contract. + /// + /// The read policy for this caller. + internal RespContext WithCachePolicy(CachePolicy policy) + => WithServices(policy ?? throw new ArgumentNullException(nameof(policy))); - /// The caller's freshness requirement, if they stated one. - internal long MaxCacheAgeTicks { get; } + /// The caller's own read policy, when they stated one; null means the cache's default. + internal CachePolicy? CachePolicy { get; } /// /// Render a command. The "" argument passes THIS CONTEXT - the receiver of the call - into the diff --git a/src/StackExchange.Redis/Contexts/RespDatabaseContext.cs b/src/StackExchange.Redis/Contexts/RespDatabaseContext.cs index 423de1b99..c313c69b5 100644 --- a/src/StackExchange.Redis/Contexts/RespDatabaseContext.cs +++ b/src/StackExchange.Redis/Contexts/RespDatabaseContext.cs @@ -87,9 +87,30 @@ namespace StackExchange.Redis /// public IConnectionMultiplexer? Multiplexer => Raw.Multiplexer; - /// A copy of this context that will not serve a cached reply older than . - /// The oldest reply this context will accept from the cache. - public RespDatabaseContext WithMaxCacheAge(TimeSpan maxAge) => new(Raw.WithMaxCacheAge(maxAge)); + /// + /// A copy of this context whose cached reads follow instead of the + /// cache's . + /// + /// + /// How this caller's reads treat cached entries: freshness tolerance, early refresh, stale-grace + /// willingness, and a removal observer (). Build it + /// once and reuse the derived context. + /// + /// + /// + /// Everything in a is applied when an entry is read, never + /// stamped when it is stored - which is exactly what makes a per-caller policy safe: the entry is + /// shared, and one copy serves callers with different tolerances. + /// + /// + /// It replaces the default policy for these reads rather than merging with it, so derive it + /// from the policy you mean: policy with { TimeToLive = ... }. And it narrows, never + /// widens: the default policy's remains a ceiling, + /// and entry lifetime and reclamation stay the cache's own business - this changes what these reads + /// will accept, never how long the shared entry lives. + /// + /// + public RespDatabaseContext WithCachePolicy(Caching.CachePolicy policy) => new(Raw.WithCachePolicy(policy)); /// A copy of this context using to remember loaded scripts. /// The script cache, or for none. diff --git a/src/StackExchange.Redis/Contexts/RespExecutor.cs b/src/StackExchange.Redis/Contexts/RespExecutor.cs index 9bab68b28..c8ec2066d 100644 --- a/src/StackExchange.Redis/Contexts/RespExecutor.cs +++ b/src/StackExchange.Redis/Contexts/RespExecutor.cs @@ -1118,7 +1118,7 @@ public static TResult Send( // not get a cached answer either, not merely that this reply is not kept if (cache is not null && UsesCache(executor, cache, flags)) { - if (TryServeFromCache(executor, ref request, handler, cache, context.MaxCacheAgeTicks, flags, out var cached)) return cached; + if (TryServeFromCache(executor, ref request, handler, cache, context.CachePolicy, flags, out var cached)) return cached; // NOTE: no in-flight wait here. Coalescing means waiting on someone else's Task, and doing // that from a synchronous caller is the sync-over-async problem this design avoids @@ -1151,7 +1151,7 @@ public static TResult Send( return default!; } - cache.TryComplete(fill, filled); + cache.TryComplete(fill, filled, context.CachePolicy); return Parse(handler, filled); } finally @@ -1230,7 +1230,7 @@ public static ValueTask SendAsync( if (cache is not null && UsesCache(executor, cache, flags)) { - if (TryServeFromCache(executor, ref request, handler, cache, context.MaxCacheAgeTicks, flags, out var cached)) + if (TryServeFromCache(executor, ref request, handler, cache, context.CachePolicy, flags, out var cached)) { return new ValueTask(cached); } @@ -1239,12 +1239,12 @@ public static ValueTask SendAsync( // copy. See design notes 6.15; this is the whole of the stampede fix at the call site. if (cache.TryAwaitInFlight(request.AsLookupKey(), executor.Database, out var pending)) { - return AwaitShared(executor, request.Detach(flags), pending, handler, cache, context.MaxCacheAgeTicks, cancellationToken); + return AwaitShared(executor, request.Detach(flags), pending, handler, cache, context.CachePolicy, cancellationToken); } if (cache.TryBeginFill(ref request, executor.Database, flags, out var fill)) { - return AwaitFill(executor, fill, handler, cache, cancellationToken); + return AwaitFill(executor, fill, handler, cache, context.CachePolicy, cancellationToken); } } @@ -1531,11 +1531,11 @@ private static bool TryServeFromCache( ref RespRequestFrame request, IRespHandler handler, RespClientCache cache, - long maxAgeTicks, + Caching.CachePolicy? policy, CommandFlags flags, [MaybeNullWhen(false)] out TResult result) { - if (!cache.TryGet(request.AsLookupKey(), executor.Database, maxAgeTicks, out var hit, out var refresh, out var reuse)) + if (!cache.TryGet(request.AsLookupKey(), executor.Database, policy, out var hit, out var refresh, out var reuse)) { result = default; return false; @@ -1706,6 +1706,7 @@ private static async ValueTask AwaitFill( RespClientCache.RespFill fill, IRespHandler handler, RespClientCache cache, + Caching.CachePolicy? policy, CancellationToken cancellationToken) { RespPayload response; @@ -1731,7 +1732,7 @@ private static async ValueTask AwaitFill( return default!; } - cache.TryComplete(fill, response); + cache.TryComplete(fill, response, policy); return Parse(handler, response); } finally @@ -1768,7 +1769,7 @@ private static async ValueTask AwaitShared( Task pending, IRespHandler handler, RespClientCache cache, - long maxAgeTicks, + Caching.CachePolicy? policy, CancellationToken cancellationToken) { try @@ -1776,7 +1777,10 @@ private static async ValueTask AwaitShared( await pending.ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); - if (cache.TryGet(owned, executor.Database, maxAgeTicks, out var hit)) + // NOTE: the fallback send below does not enroll the policy's removal observer - it has no + // fill, so no captured dependencies. Rare (the leader's reply was not cacheable), and the + // consumer's own freshness bound covers it exactly as it covers an uncached read. + if (cache.TryGet(owned, executor.Database, policy, out var hit, out _, out _)) { try { diff --git a/src/StackExchange.Redis/Contexts/RespServerContext.cs b/src/StackExchange.Redis/Contexts/RespServerContext.cs index cfd41e2e7..9fd6da875 100644 --- a/src/StackExchange.Redis/Contexts/RespServerContext.cs +++ b/src/StackExchange.Redis/Contexts/RespServerContext.cs @@ -75,9 +75,8 @@ namespace StackExchange.Redis /// public IConnectionMultiplexer? Multiplexer => Raw.Multiplexer; - /// A copy of this context that will not serve a cached reply older than . - /// The oldest reply this context will accept from the cache. - public RespServerContext WithMaxCacheAge(TimeSpan maxAge) => new(Raw.WithMaxCacheAge(maxAge)); + /// + public RespServerContext WithCachePolicy(Caching.CachePolicy policy) => new(Raw.WithCachePolicy(policy)); /// A copy of this context using to remember loaded scripts. /// The script cache, or for none. diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index ca33ae50e..e7093f5c8 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -37,6 +37,15 @@ StackExchange.Redis.Caching.CacheAdmission StackExchange.Redis.Caching.CacheAdmission.Default = 0 -> StackExchange.Redis.Caching.CacheAdmission StackExchange.Redis.Caching.CacheAdmission.OnFirstMiss = 1 -> StackExchange.Redis.Caching.CacheAdmission StackExchange.Redis.Caching.CacheAdmission.OnRepeatedMiss = 2 -> StackExchange.Redis.Caching.CacheAdmission +StackExchange.Redis.Caching.CacheKeyRemovalReason +StackExchange.Redis.Caching.CacheKeyRemovalReason.Invalidated = 0 -> StackExchange.Redis.Caching.CacheKeyRemovalReason +StackExchange.Redis.Caching.CacheKeyRemovalReason.LocalWrite = 1 -> StackExchange.Redis.Caching.CacheKeyRemovalReason +StackExchange.Redis.Caching.CacheKeyRemovalReason.Replaced = 2 -> StackExchange.Redis.Caching.CacheKeyRemovalReason +StackExchange.Redis.Caching.CacheKeyRemovalReason.Expired = 3 -> StackExchange.Redis.Caching.CacheKeyRemovalReason +StackExchange.Redis.Caching.CacheKeyRemovalReason.Evicted = 4 -> StackExchange.Redis.Caching.CacheKeyRemovalReason +StackExchange.Redis.Caching.CacheKeyRemovalReason.Flushed = 5 -> StackExchange.Redis.Caching.CacheKeyRemovalReason +StackExchange.Redis.Caching.CacheKeyRemovedHandler +virtual StackExchange.Redis.Caching.CacheKeyRemovedHandler.Invoke(System.ReadOnlySpan key, StackExchange.Redis.Caching.CacheKeyRemovalReason reason) -> void StackExchange.Redis.Caching.CacheOptions.Admission.get -> StackExchange.Redis.Caching.CacheAdmission StackExchange.Redis.Caching.CacheOptions.Admission.init -> void StackExchange.Redis.Caching.CacheOptions @@ -62,11 +71,20 @@ StackExchange.Redis.Caching.CacheOptions.TrackingMode.get -> StackExchange.Redis StackExchange.Redis.Caching.CacheOptions.TrackingMode.init -> void StackExchange.Redis.Caching.CachePolicy StackExchange.Redis.Caching.CachePolicy.CachePolicy() -> void +StackExchange.Redis.Caching.CachePolicy.$() -> StackExchange.Redis.Caching.CachePolicy! +StackExchange.Redis.Caching.CachePolicy.Equals(StackExchange.Redis.Caching.CachePolicy? other) -> bool +override StackExchange.Redis.Caching.CachePolicy.Equals(object? obj) -> bool +override StackExchange.Redis.Caching.CachePolicy.GetHashCode() -> int +override StackExchange.Redis.Caching.CachePolicy.ToString() -> string! +static StackExchange.Redis.Caching.CachePolicy.operator !=(StackExchange.Redis.Caching.CachePolicy? left, StackExchange.Redis.Caching.CachePolicy? right) -> bool +static StackExchange.Redis.Caching.CachePolicy.operator ==(StackExchange.Redis.Caching.CachePolicy? left, StackExchange.Redis.Caching.CachePolicy? right) -> bool static StackExchange.Redis.Caching.CachePolicy.Default.get -> StackExchange.Redis.Caching.CachePolicy! StackExchange.Redis.Caching.CachePolicy.InvalidationGracePeriod.get -> System.TimeSpan StackExchange.Redis.Caching.CachePolicy.InvalidationGracePeriod.init -> void StackExchange.Redis.Caching.CachePolicy.RefreshAfter.get -> System.TimeSpan StackExchange.Redis.Caching.CachePolicy.RefreshAfter.init -> void +StackExchange.Redis.Caching.CachePolicy.RemovalHandler.get -> StackExchange.Redis.Caching.CacheKeyRemovedHandler? +StackExchange.Redis.Caching.CachePolicy.RemovalHandler.init -> void StackExchange.Redis.Caching.CachePolicy.TimeToLive.get -> System.TimeSpan StackExchange.Redis.Caching.CachePolicy.TimeToLive.init -> void StackExchange.Redis.Caching.CacheTrackingMode @@ -449,7 +467,7 @@ StackExchange.Redis.RespDatabaseContext.Database.get -> int StackExchange.Redis.RespDatabaseContext.RespDatabaseContext() -> void StackExchange.Redis.RespDatabaseContext.RespDatabaseContext(StackExchange.Redis.RespContext! context) -> void StackExchange.Redis.RespDatabaseContext.WithDatabase(int database) -> StackExchange.Redis.RespDatabaseContext -StackExchange.Redis.RespDatabaseContext.WithMaxCacheAge(System.TimeSpan maxAge) -> StackExchange.Redis.RespDatabaseContext +StackExchange.Redis.RespDatabaseContext.WithCachePolicy(StackExchange.Redis.Caching.CachePolicy! policy) -> StackExchange.Redis.RespDatabaseContext StackExchange.Redis.RespDatabaseContext.WithServices(object? services) -> StackExchange.Redis.RespDatabaseContext StackExchange.Redis.RespDatabaseContext.WithoutCache() -> StackExchange.Redis.RespDatabaseContext static StackExchange.Redis.RespDatabaseContext.explicit operator StackExchange.Redis.RespContext!(StackExchange.Redis.RespDatabaseContext context) -> StackExchange.Redis.RespContext! @@ -560,8 +578,8 @@ StackExchange.Redis.RespServerContext.Blocking() -> StackExchange.Redis.RespServ StackExchange.Redis.RespServerContext.Multiplexer.get -> StackExchange.Redis.IConnectionMultiplexer? StackExchange.Redis.RespServerContext.RespServerContext() -> void StackExchange.Redis.RespServerContext.RespServerContext(StackExchange.Redis.RespContext! context) -> void -StackExchange.Redis.RespServerContext.WithMaxCacheAge(System.TimeSpan maxAge) -> StackExchange.Redis.RespServerContext StackExchange.Redis.RespServerContext.WithServices(object? services) -> StackExchange.Redis.RespServerContext +StackExchange.Redis.RespServerContext.WithCachePolicy(StackExchange.Redis.Caching.CachePolicy! policy) -> StackExchange.Redis.RespServerContext StackExchange.Redis.RespServerContext.WithoutCache() -> StackExchange.Redis.RespServerContext static StackExchange.Redis.RespServerContext.explicit operator StackExchange.Redis.RespContext!(StackExchange.Redis.RespServerContext context) -> StackExchange.Redis.RespContext! static StackExchange.Redis.RespServerContext.explicit operator StackExchange.Redis.RespContext(StackExchange.Redis.RespServerContext context) -> StackExchange.Redis.RespContext diff --git a/tests/StackExchange.Redis.Tests/RespCacheLifetimeTests.cs b/tests/StackExchange.Redis.Tests/RespCacheLifetimeTests.cs index 5735d0949..2e127c218 100644 --- a/tests/StackExchange.Redis.Tests/RespCacheLifetimeTests.cs +++ b/tests/StackExchange.Redis.Tests/RespCacheLifetimeTests.cs @@ -91,12 +91,12 @@ public async Task AnExpiredEntryIsReplacedByTheNextFill() [Fact] public async Task AContextCanDemandSomethingFresherThanThePolicy() { - // the one knob that is per-call, because freshness tolerance is a property of the caller - and the - // one that could not be added to IDatabase at all without a binary break + // the policy is per-call, because freshness tolerance is a property of the caller - and the + // thing that could not be added to IDatabase at all without a binary break using var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss, DefaultPolicy = new CachePolicy { TimeToLive = TimeSpan.FromHours(1) } }); var executor = new FakeExecutor("$1\r\na\r\n", "$1\r\nb\r\n"); var relaxed = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache)); - var picky = relaxed.WithMaxCacheAge(TimeSpan.FromMilliseconds(50)); + var picky = relaxed.WithCachePolicy(new CachePolicy { TimeToLive = TimeSpan.FromMilliseconds(50) }); Assert.Equal("a", await Get(relaxed)); await Task.Delay(150); @@ -118,7 +118,7 @@ public async Task OneEntryServesCallersWithDifferentTolerances() using var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss, DefaultPolicy = new CachePolicy { TimeToLive = TimeSpan.FromHours(1) } }); var executor = new FakeExecutor("$1\r\na\r\n"); var relaxed = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache)); - var picky = relaxed.WithMaxCacheAge(TimeSpan.FromMinutes(30)); + var picky = relaxed.WithCachePolicy(new CachePolicy { TimeToLive = TimeSpan.FromMinutes(30) }); Assert.Equal("a", await Get(relaxed)); Assert.Equal("a", await Get(picky)); @@ -134,7 +134,7 @@ public async Task AContextCannotAskForStalerThanThePolicyAllows() using var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss, DefaultPolicy = new CachePolicy { TimeToLive = TimeSpan.FromMilliseconds(80) } }); var executor = new FakeExecutor("$1\r\na\r\n", "$1\r\nb\r\n"); var context = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache) - .WithMaxCacheAge(TimeSpan.FromHours(1))); + .WithCachePolicy(new CachePolicy { TimeToLive = TimeSpan.FromHours(1) })); Assert.Equal("a", await Get(context)); await Task.Delay(200); diff --git a/tests/StackExchange.Redis.Tests/RespCacheRemovalTests.cs b/tests/StackExchange.Redis.Tests/RespCacheRemovalTests.cs new file mode 100644 index 000000000..3e3eb13da --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RespCacheRemovalTests.cs @@ -0,0 +1,465 @@ +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using RESPite.Messages; +using StackExchange.Redis.Caching; +using StackExchange.Redis.Protocol; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Per-key removal notification: a - per caller via +/// , or connection-wide on the default policy - is +/// told when a key read under it stops being servable from the client-side cache. +/// +/// +/// +/// The contract under test: enrollment is armed by a serve (hit or fill) and consumed by the first +/// removal at or after it - one notification per enrollment, re-armed by reading again. It fires at +/// invalidation time (the entry may still be physically resident), from the thread pool - so every test +/// collects events through and awaits them rather than asserting inline. +/// +/// +/// Deaths that cannot be announced are not enrolled at all (a fill whose tracking prefix was dropped), +/// and a value handed out already-dead - a fill that raced an invalidation, a stale-grace serve - fires +/// immediately rather than waiting for a removal that already happened. +/// +/// +public class RespCacheRemovalTests +{ + private const CommandFlags Readable = CommandFlags.CommandRetryReadOnly; + + private static readonly RespContext Renderer = new(); + + private static (RespDatabaseContext Context, RespClientCache Cache, FakeExecutor Executor, Recorder Recorder) Create( + CacheOptions? options = null, + params string[] replies) + { + options ??= new CacheOptions { Admission = CacheAdmission.OnFirstMiss }; + var cache = new RespClientCache(options); + var executor = new FakeExecutor(replies); + var recorder = new Recorder(); + + // the caller's read policy is the cache's own, plus the observer: derive, don't rebuild + var context = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache)) + .WithCachePolicy(options.DefaultPolicy with { RemovalHandler = recorder.Handler }); + return (context, cache, executor, recorder); + } + + [Fact] + public async Task AForeignInvalidationFiresWithTheKeyAndReason_WhileTheEntryIsStillResident() + { + var (context, cache, _, recorder) = Create(replies: "$1\r\na\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + + cache.OnInvalidate("k"u8); + + var (key, reason) = await recorder.NextAsync(); + Assert.Equal("k", key); + Assert.Equal(CacheKeyRemovalReason.Invalidated, reason); + + // fired at INVALIDATION time: the corpse is still in the dictionary, nothing swept it + Assert.Equal(1, cache.Count); + await recorder.AssertNoFurtherAsync(); + } + } + + [Fact] + public async Task OurOwnWriteSaysLocalWrite() + { + var (context, cache, _, recorder) = Create(null, "$1\r\na\r\n", "+OK\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + _ = await context.SendAsync($"{RedisCommand.SET}{(RedisKey)"k"}{(RedisValue)"b"}", CommandFlags.None); + + var (key, reason) = await recorder.NextAsync(); + Assert.Equal("k", key); + Assert.Equal(CacheKeyRemovalReason.LocalWrite, reason); + await recorder.AssertNoFurtherAsync(); + } + } + + [Fact] + public async Task OneEnrollmentFiresOnce_AndReadingAgainReArms() + { + var (context, cache, _, recorder) = Create(replies: "$1\r\na\r\n"); + using (cache) + { + // several reads, ONE enrollment: arming is idempotent per handler instance + for (var i = 0; i < 3; i++) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + } + + cache.OnInvalidate("k"u8); + Assert.Equal(CacheKeyRemovalReason.Invalidated, (await recorder.NextAsync()).Reason); + + // a second invalidation of the dead key says nothing: the first one consumed the enrollment + cache.OnInvalidate("k"u8); + await recorder.AssertNoFurtherAsync(); + + // reading again re-arms, and the next removal is announced again + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + cache.OnInvalidate("k"u8); + Assert.Equal(CacheKeyRemovalReason.Invalidated, (await recorder.NextAsync()).Reason); + await recorder.AssertNoFurtherAsync(); + } + } + + [Fact] + public async Task AFillThatRacedAnInvalidationFiresImmediately() + { + // the invalidation lands between the fill's generation capture and its completion: the store is + // refused, and the consumer - who was just handed a value of unprovable currency - is told now, + // because the removal it would otherwise wait for has already happened + var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss }); + var executor = new InvalidatingExecutor(cache, "k", "$1\r\na\r\n"); + var recorder = new Recorder(); + var context = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache)) + .WithCachePolicy(new CachePolicy { RemovalHandler = recorder.Handler }); + using (cache) + { + var value = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + Assert.Equal("a", value); // served to the caller regardless + + var (key, reason) = await recorder.NextAsync(); + Assert.Equal("k", key); + Assert.Equal(CacheKeyRemovalReason.Invalidated, reason); + Assert.Equal(0, cache.Count); // and indeed, nothing was stored + await recorder.AssertNoFurtherAsync(); + } + } + + [Fact] + public async Task AReplacingRefreshSaysReplaced() + { + // refresh-ahead: a read past RefreshAfter re-fetches in the background, and the completed + // refresh REPLACES the live entry - the one removal that happens with no invalidation at all + var options = new CacheOptions + { + Admission = CacheAdmission.OnFirstMiss, + DefaultPolicy = new CachePolicy { RefreshAfter = TimeSpan.FromMilliseconds(1) }, + }; + var (context, cache, _, recorder) = Create(options, "$1\r\na\r\n", "$1\r\nb\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + await Task.Delay(50); // comfortably past RefreshAfter, well inside TimeToLive + + // this hit is served the old value AND claims the refresh that will replace it + Assert.Equal("a", await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable)); + + var (key, reason) = await recorder.NextAsync(); + Assert.Equal("k", key); + Assert.Equal(CacheKeyRemovalReason.Replaced, reason); + await recorder.AssertNoFurtherAsync(); + } + } + + [Fact] + public async Task EvictionUnderBudgetSaysEvicted() + { + var options = new CacheOptions { Admission = CacheAdmission.OnFirstMiss, MaxEntries = 1 }; + var (context, cache, _, recorder) = Create(options, "$1\r\na\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k1"}", Readable); + + // the second entry pushes the first out: nothing wrong with it, the budget just ran out - + // and the server may still track it, so the local death must be announced, not inferred + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k2"}", Readable); + + var (key, reason) = await recorder.NextAsync(); + Assert.Equal("k1", key); + Assert.Equal(CacheKeyRemovalReason.Evicted, reason); + } + } + + [Fact] + public async Task AnEntryThatAgesOutSaysExpiredWhenReclaimed() + { + var options = new CacheOptions + { + Admission = CacheAdmission.OnFirstMiss, + DefaultPolicy = new CachePolicy { TimeToLive = TimeSpan.FromMilliseconds(1) }, + }; + var (context, cache, _, recorder) = Create(options, "$1\r\na\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + await Task.Delay(50); // comfortably past the lifetime + + Assert.Equal(1, cache.Sweep()); // lazy expiry: the sweep is the reclamation moment + var (key, reason) = await recorder.NextAsync(); + Assert.Equal("k", key); + Assert.Equal(CacheKeyRemovalReason.Expired, reason); + await recorder.AssertNoFurtherAsync(); + } + } + + [Fact] + public async Task AFlushSaysFlushedPromptly() + { + var (context, cache, _, recorder) = Create(replies: "$1\r\na\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + + cache.OnFlush(); // reconnect, FLUSHALL, tracking torn down - no sweep involved + + var (key, reason) = await recorder.NextAsync(); + Assert.Equal("k", key); + Assert.Equal(CacheKeyRemovalReason.Flushed, reason); + await recorder.AssertNoFurtherAsync(); + } + } + + [Fact] + public async Task TwoConsumersOnOneKeyBothHear() + { + var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss }); + var executor = new FakeExecutor("$1\r\na\r\n"); + var shared = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache)); + var first = new Recorder(); + var second = new Recorder(); + using (cache) + { + _ = await shared.WithCachePolicy(new CachePolicy { RemovalHandler = first.Handler }).SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + _ = await shared.WithCachePolicy(new CachePolicy { RemovalHandler = second.Handler }).SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + + cache.OnInvalidate("k"u8); + + Assert.Equal(("k", CacheKeyRemovalReason.Invalidated), await first.NextAsync()); + Assert.Equal(("k", CacheKeyRemovalReason.Invalidated), await second.NextAsync()); + } + } + + [Fact] + public async Task AThrowingConsumerDoesNotSilenceAnother() + { + var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss }); + var executor = new FakeExecutor("$1\r\na\r\n"); + var shared = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache)); + CacheKeyRemovedHandler throwing = (_, _) => throw new InvalidOperationException("deliberately rude"); + var recorder = new Recorder(); + using (cache) + { + // the throwing handler enrolls FIRST, so a propagated exception would take the second with it + _ = await shared.WithCachePolicy(new CachePolicy { RemovalHandler = throwing }).SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + _ = await shared.WithCachePolicy(new CachePolicy { RemovalHandler = recorder.Handler }).SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + + cache.OnInvalidate("k"u8); + + Assert.Equal(("k", CacheKeyRemovalReason.Invalidated), await recorder.NextAsync()); + } + } + + [Fact] + public async Task AHitEnrollsAnEntryThatWasFilledThroughAPlainContext() + { + var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss }); + var executor = new FakeExecutor("$1\r\na\r\n"); + var plain = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache)); + var recorder = new Recorder(); + using (cache) + { + _ = await plain.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); // fill, no observer + _ = await plain.WithCachePolicy(new CachePolicy { RemovalHandler = recorder.Handler }).SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); // hit enrolls + + cache.OnInvalidate("k"u8); + + Assert.Equal(("k", CacheKeyRemovalReason.Invalidated), await recorder.NextAsync()); + await recorder.AssertNoFurtherAsync(); + } + } + + [Fact] + public async Task AHandlerOnTheDefaultPolicyIsConnectionWide() + { + // no per-context policy anywhere: the observer rides the cache's own DefaultPolicy, so EVERY + // read enrolls it - the firehose a backplane wants, with the same read-armed one-shot contract + var recorder = new Recorder(); + var cache = new RespClientCache(new CacheOptions + { + Admission = CacheAdmission.OnFirstMiss, + DefaultPolicy = new CachePolicy { RemovalHandler = recorder.Handler }, + }); + var executor = new FakeExecutor("$1\r\na\r\n"); + var plain = new RespDatabaseContext(new RespContext().WithExecutor(executor).WithCache(cache)); + using (cache) + { + _ = await plain.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + + cache.OnInvalidate("k"u8); + + Assert.Equal(("k", CacheKeyRemovalReason.Invalidated), await recorder.NextAsync()); + await recorder.AssertNoFurtherAsync(); + } + } + + [Fact] + public async Task TheKeyArrivesExactlyAsStored_PrefixesIncluded() + { + var (context, cache, _, recorder) = Create(replies: "$1\r\na\r\n"); + using (cache) + { + _ = await context.AppendKeyPrefix("p:").SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + + cache.OnInvalidate("p:k"u8); // the server names the rendered key, and so does the notification + + var (key, reason) = await recorder.NextAsync(); + Assert.Equal("p:k", key); + Assert.Equal(CacheKeyRemovalReason.Invalidated, reason); + } + } + + [Fact] + public async Task AStaleGraceServeIsAnnouncedImmediately() + { + var options = new CacheOptions + { + Admission = CacheAdmission.OnFirstMiss, + DefaultPolicy = new CachePolicy { InvalidationGracePeriod = TimeSpan.FromSeconds(30) }, + }; + var (context, cache, _, recorder) = Create(options, "$1\r\na\r\n"); + using (cache) + { + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + + cache.OnInvalidate("k"u8); + Assert.Equal(CacheKeyRemovalReason.Invalidated, (await recorder.NextAsync()).Reason); + + // within the grace window the dead value is still served - and the serve says so at once, + // rather than handing a non-current value over in silence + _ = await context.SendAsync($"{RedisCommand.GET}{(RedisKey)"k"}", Readable); + Assert.Equal(CacheKeyRemovalReason.Invalidated, (await recorder.NextAsync()).Reason); + } + } + + [Fact] + public void SlabCompactionIsNotARemoval_AndTheSurvivingEnrollmentStillFires() + { + // the slab-churn setup from AReadEntryIsCarriedForwardWhenItsSlabIsEvicted: a hot entry is + // carried forward (CopyForward) when its slab is evicted. Enrollment lives on the key's NODE, + // so the move must say nothing - and the enrollment must still be armed afterwards + const long budget = 4L * 1024 * 1024; + var reply = "$8000\r\n" + new string('x', 8000) + "\r\n"; + var recorder = new Recorder(); + var observing = new CachePolicy { RemovalHandler = recorder.Handler }; + using var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss, MaxBytes = budget }); + + Fill(cache, "hot", reply, observing); + for (var i = 0; i < 10_000 && cache.Promoted == 0; i++) + { + Fill(cache, "filler" + i, reply); + if (i % 4 == 0) + { + using var frame = Renderer.Render($"{RedisCommand.GET}{(RedisKey)"hot"}"); + Assert.True(cache.TryGet(frame.AsLookupKey(Readable), 0, observing, out var payload, out _, out _)); + payload.Release(); + } + } + + Assert.True(cache.Promoted > 0, "the hot entry was never carried forward"); + recorder.AssertNothingYet(); // the move is not a removal + + cache.OnInvalidate("hot"u8); + var (key, reason) = recorder.Next(); + Assert.Equal("hot", key); + Assert.Equal(CacheKeyRemovalReason.Invalidated, reason); + } + + /// Render, fill, and complete - optionally under an observing policy, as the funnel would. + private static void Fill(RespClientCache cache, string key, string response, CachePolicy? policy = null) + { + var frame = Renderer.Render($"{RedisCommand.GET}{(RedisKey)key}"); + if (!cache.TryBeginFill(ref frame, 0, Readable, out var fill)) + { + frame.Dispose(); + return; + } + + Complete(cache, fill, response, policy); + } + + private static void Complete(RespClientCache cache, in RespClientCache.RespFill fill, string response, CachePolicy? policy = null) + { + var payload = RespPayload.Create(Encoding.UTF8.GetBytes(response)); + try + { + cache.TryComplete(fill, payload, policy); + } + finally + { + payload.Release(); + } + } + + /// Invalidates the key after every send: a write landing while the fill is in flight. + private sealed class InvalidatingExecutor(RespClientCache cache, string key, params string[] replies) : FakeExecutor(replies) + { + protected override void OnSent(in RespRequest request) => cache.OnInvalidate(Encoding.UTF8.GetBytes(key)); + } + + /// + /// Collects notifications - which arrive from the thread pool - so tests can await them. + /// + private sealed class Recorder + { + private readonly ConcurrentQueue<(string Key, CacheKeyRemovalReason Reason)> _events = new(); + private readonly SemaphoreSlim _signal = new(0); + + public Recorder() + { + // ONE delegate instance for the recorder's lifetime: enrollment dedupes on its identity + Handler = (key, reason) => + { + _events.Enqueue((Encoding.UTF8.GetString(key.ToArray()), reason)); + _signal.Release(); + }; + } + + public CacheKeyRemovedHandler Handler { get; } + + public async Task<(string Key, CacheKeyRemovalReason Reason)> NextAsync() + { + Assert.True(await _signal.WaitAsync(TimeSpan.FromSeconds(10)), "timed out waiting for a removal notification"); + Assert.True(_events.TryDequeue(out var fired)); + return fired; + } + + public (string Key, CacheKeyRemovalReason Reason) Next() + { + Assert.True(_signal.Wait(TimeSpan.FromSeconds(10)), "timed out waiting for a removal notification"); + Assert.True(_events.TryDequeue(out var fired)); + return fired; + } + + /// Settle the pool briefly and assert no stray notification arrived. + public async Task AssertNoFurtherAsync() + { + if (await _signal.WaitAsync(TimeSpan.FromMilliseconds(250))) + { + _events.TryDequeue(out var fired); + Assert.Fail($"unexpected notification: {fired}"); + } + } + + /// As , for a synchronous test. + public void AssertNothingYet() + { + if (_signal.Wait(TimeSpan.FromMilliseconds(250))) + { + _events.TryDequeue(out var fired); + Assert.Fail($"unexpected notification: {fired}"); + } + } + } +} diff --git a/tests/StackExchange.Redis.Tests/RespStaleWhileRevalidateTests.cs b/tests/StackExchange.Redis.Tests/RespStaleWhileRevalidateTests.cs index d2e6a45e3..ddfc723b6 100644 --- a/tests/StackExchange.Redis.Tests/RespStaleWhileRevalidateTests.cs +++ b/tests/StackExchange.Redis.Tests/RespStaleWhileRevalidateTests.cs @@ -258,13 +258,14 @@ public async Task TheGracePeriodNeverOutlivesTimeToLive() [Fact] public async Task TheGracePeriodHonoursTheCallersMaxAge() { - // WithMaxCacheAge is the caller's own freshness requirement; the grace period is the POLICY's + // a read policy's TimeToLive is the caller's own freshness requirement; the grace period is a // willingness to serve stale, and cannot overrule a caller who asked for fresher - using var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss, DefaultPolicy = new CachePolicy + var policy = new CachePolicy { InvalidationGracePeriod = TimeSpan.FromSeconds(5), TimeToLive = TimeSpan.FromMinutes(5), - } }); + }; + using var cache = new RespClientCache(new CacheOptions { Admission = CacheAdmission.OnFirstMiss, DefaultPolicy = policy }); var executor = new FakeExecutor("$1\r\na\r\n", "$1\r\nb\r\n"); var context = Context(executor, cache); @@ -272,7 +273,7 @@ public async Task TheGracePeriodHonoursTheCallersMaxAge() await Task.Delay(200); cache.OnInvalidate(Encoding.UTF8.GetBytes("k")); - var strict = context.WithMaxCacheAge(TimeSpan.FromMilliseconds(100)); + var strict = context.WithCachePolicy(policy with { TimeToLive = TimeSpan.FromMilliseconds(100) }); Assert.Equal("b", await Get(strict)); Assert.Equal(0, cache.ServedStale); Assert.Equal(2, executor.Sends);