diff --git a/Stampede.Http.Tests/Caching/BackgroundRevalidationCoordinatorTests.cs b/Stampede.Http.Tests/Caching/BackgroundRevalidationCoordinatorTests.cs new file mode 100644 index 0000000..ed4c3ff --- /dev/null +++ b/Stampede.Http.Tests/Caching/BackgroundRevalidationCoordinatorTests.cs @@ -0,0 +1,107 @@ +using Stampede.Http.Caching; +using FluentAssertions; + +namespace Stampede.Http.Tests.Caching; + +/// +/// Verifies the stale-while-revalidate deduplication contract (RFC 5861 §3): a key has at most one +/// background refresh in flight, and the claim is always released once that refresh finishes. +/// +public sealed class BackgroundRevalidationCoordinatorTests +{ + [Fact] + public async Task Schedule_WhileRefreshInFlight_DoesNotStartASecondOne() + { + BackgroundRevalidationCoordinator coordinator = new(); + TaskCompletionSource gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + int started = 0; + + for (int i = 0; i < 10; i++) + { + coordinator.Schedule("key", async () => + { + _ = Interlocked.Increment(ref started); + await gate.Task; + }); + } + + // Give the scheduled work a chance to run before asserting. + await Task.Delay(100, TestContext.Current.CancellationToken); + + Volatile.Read(ref started).Should().Be(1, "a key already being revalidated must not be revalidated again"); + + gate.SetResult(); + } + + [Fact] + public async Task Schedule_DifferentKeys_RunIndependently() + { + BackgroundRevalidationCoordinator coordinator = new(); + TaskCompletionSource gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + int started = 0; + + for (int i = 0; i < 5; i++) + { + coordinator.Schedule($"key-{i}", async () => + { + _ = Interlocked.Increment(ref started); + await gate.Task; + }); + } + + await Task.Delay(100, TestContext.Current.CancellationToken); + + Volatile.Read(ref started).Should().Be(5, "deduplication is per key, not global"); + + gate.SetResult(); + } + + [Fact] + public async Task Schedule_AfterPreviousRefreshCompletes_StartsAgain() + { + BackgroundRevalidationCoordinator coordinator = new(); + int started = 0; + + for (int i = 0; i < 20; i++) + { + coordinator.Schedule("key", () => + { + _ = Interlocked.Increment(ref started); + return Task.CompletedTask; + }); + + // Let the refresh finish and release its claim before scheduling the next one. + await Task.Delay(20, TestContext.Current.CancellationToken); + } + + // Regression: claiming the key from inside the background task allowed the refresh to finish before + // its own registration landed, leaving the key claimed forever and blocking all later revalidations. + Volatile.Read(ref started).Should().Be(20, + "each completed refresh must release its key so the entry can be revalidated again"); + } + + [Fact] + public async Task Schedule_WhenRefreshThrows_ReleasesTheKey() + { + BackgroundRevalidationCoordinator coordinator = new(); + int started = 0; + + coordinator.Schedule("key", () => + { + _ = Interlocked.Increment(ref started); + throw new InvalidOperationException("origin unreachable"); + }); + + await Task.Delay(50, TestContext.Current.CancellationToken); + + coordinator.Schedule("key", () => + { + _ = Interlocked.Increment(ref started); + return Task.CompletedTask; + }); + + await Task.Delay(50, TestContext.Current.CancellationToken); + + Volatile.Read(ref started).Should().Be(2, "a failed refresh must not block the key permanently"); + } +} diff --git a/Stampede.Http.Tests/Caching/HeadMethodCachingTests.cs b/Stampede.Http.Tests/Caching/HeadMethodCachingTests.cs index 2eba266..9a1c1bb 100644 --- a/Stampede.Http.Tests/Caching/HeadMethodCachingTests.cs +++ b/Stampede.Http.Tests/Caching/HeadMethodCachingTests.cs @@ -77,6 +77,61 @@ public async Task Head_FromCache_ResponseBodyIsEmpty() body.Should().BeEmpty("HEAD responses must not include a body"); } + [Fact] + public async Task Head_FromCache_PreservesContentHeaders() + { + (CachingMiddleware middleware, _) = BuildPipeline(_ => + new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent("body text", System.Text.Encoding.UTF8, "application/json") + }); + + HttpMessageInvoker invoker = Invoker(middleware); + const string url = "https://api.test/head/content-headers"; + + HttpResponseMessage getResponse = await invoker.SendAsync(new HttpRequestMessage(HttpMethod.Get, url), CancellationToken.None); + HttpResponseMessage headResponse = await invoker.SendAsync(new HttpRequestMessage(HttpMethod.Head, url), CancellationToken.None); + + // RFC 9110 §9.3.2 — HEAD repeats the header fields the equivalent GET would have sent + headResponse.Content.Headers.ContentType?.MediaType.Should().Be("application/json", + "a cached HEAD must repeat the Content-Type the equivalent GET would have sent"); + headResponse.Content.Headers.ContentLength.Should().Be(getResponse.Content.Headers.ContentLength, + "a cached HEAD must report the same Content-Length as the equivalent GET"); + } + + [Fact] + public async Task Head_AfterRevalidation_PreservesContentHeaders() + { + int callCount = 0; + (CachingMiddleware middleware, _) = BuildPipeline(_ => + { + callCount++; + if (callCount == 1) + { + HttpResponseMessage r = new(HttpStatusCode.OK) + { + Content = new StringContent("data", System.Text.Encoding.UTF8, "application/json") + }; + r.Headers.ETag = new EntityTagHeaderValue("\"v1\""); + r.Headers.CacheControl = new CacheControlHeaderValue { MaxAge = TimeSpan.Zero }; + return r; + } + + return new HttpResponseMessage(HttpStatusCode.NotModified); + }); + + HttpMessageInvoker invoker = Invoker(middleware); + const string url = "https://api.test/head/reval-content-headers"; + + _ = await invoker.SendAsync(new HttpRequestMessage(HttpMethod.Get, url), CancellationToken.None); + HttpResponseMessage headResponse = await invoker.SendAsync(new HttpRequestMessage(HttpMethod.Head, url), CancellationToken.None); + + headResponse.Content.Headers.ContentType?.MediaType.Should().Be("application/json", + "a HEAD refreshed by a 304 must still repeat the stored Content-Type"); + byte[] body = await headResponse.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + body.Should().BeEmpty("HEAD response body must be empty even after revalidation"); + } + [Fact] public async Task Head_FromCache_AgeHeaderPresent() { diff --git a/Stampede.Http.Tests/Caching/MemoryCacheStoreExpirationTests.cs b/Stampede.Http.Tests/Caching/MemoryCacheStoreExpirationTests.cs index 1e136ac..653b05f 100644 --- a/Stampede.Http.Tests/Caching/MemoryCacheStoreExpirationTests.cs +++ b/Stampede.Http.Tests/Caching/MemoryCacheStoreExpirationTests.cs @@ -39,17 +39,44 @@ private static CacheEntry BuildEntry( // ── AbsoluteExpiration placement ───────────────────────────────────────── [Fact] - public void Set_WithoutStaleWindow_PastExpiry_EntryStaysForConditionalRevalidation() + public void Set_WithoutStaleWindow_PastExpiry_NoValidator_EntryNotRetained() { MemoryCacheStore store = CreateStore(); - // ExpiresAt in the past and no stale window → evictionTtl ≤ 0 → no AbsoluteExpiration set. - // Entry must remain so its ETag / Last-Modified can be used for conditional revalidation. + // ExpiresAt in the past, no stale window and no validator → the entry can neither be served + // (it is already stale) nor revalidated (nothing to condition on), so it must not be retained. CacheEntry entry = BuildEntry(expiresAt: DateTimeOffset.UtcNow.AddSeconds(-1)); store.Set("key1", entry); bool found = store.TryGetValue("key1", out _); - found.Should().BeTrue("past-ExpiresAt with no stale window should stay for conditional revalidation (LRU eviction only)"); + found.Should().BeFalse("an entry that can never be served nor revalidated must not occupy the cache"); + } + + [Fact] + public void Set_WithoutStaleWindow_PastExpiry_WithValidator_RetainedByGrace() + { + MemoryCacheStore store = CreateStore(); + // Same as above but with an ETag: the revalidation grace period keeps it available so the next + // request can send a conditional If-None-Match instead of a full refetch. + CacheEntry entry = BuildEntry(expiresAt: DateTimeOffset.UtcNow.AddSeconds(-1), eTag: "\"v1\""); + + store.Set("key1-validator", entry); + + bool found = store.TryGetValue("key1-validator", out _); + found.Should().BeTrue("a validator-carrying entry is retained by RevalidationGraceSeconds for conditional revalidation"); + } + + [Fact] + public void Set_UnusableEntry_RemovesPreviousRepresentation() + { + MemoryCacheStore store = CreateStore(); + store.Set("superseded", BuildEntry(expiresAt: DateTimeOffset.UtcNow.AddMinutes(5))); + + // Storing an unusable representation must not silently leave the previous one behind. + store.Set("superseded", BuildEntry(expiresAt: DateTimeOffset.UtcNow.AddSeconds(-1))); + + bool found = store.TryGetValue("superseded", out _); + found.Should().BeFalse("a superseded representation must not keep being served after an unusable store"); } [Fact] @@ -99,11 +126,11 @@ public void Set_LargerStaleWindow_UsedAsDeadline() } [Fact] - public void Set_BothStaleWindowsZero_PastExpiry_EntryStaysForConditionalRevalidation() + public void Set_BothStaleWindowsZero_PastExpiry_NoValidator_EntryNotRetained() { MemoryCacheStore store = CreateStore(); - // Both stale windows are zero and ExpiresAt is in the past → evictionTtl ≤ 0 → no expiration - // set on the IMemoryCache entry. Entry must remain for conditional revalidation. + // Both stale windows are zero, ExpiresAt is in the past and there is no validator → the entry has + // no usable window at all and must not be retained. CacheEntry entry = BuildEntry( expiresAt: DateTimeOffset.UtcNow.AddSeconds(-1), staleIfErrorSeconds: 0, @@ -112,7 +139,34 @@ public void Set_BothStaleWindowsZero_PastExpiry_EntryStaysForConditionalRevalida store.Set("key5", entry); bool found = store.TryGetValue("key5", out _); - found.Should().BeTrue("past-ExpiresAt with no stale windows should stay for conditional revalidation (LRU eviction only)"); + found.Should().BeFalse("an entry with no freshness, no stale window and no validator must not occupy the cache"); + } + + [Fact] + public void Set_MaxAgeZeroWithoutValidator_NotRetainedEvenWithoutSizeLimit() + { + // Regression: without MaxCacheSize the IMemoryCache has no SizeLimit and therefore no LRU eviction, + // so an entry stored with no expiration would live for the lifetime of the process. + MemoryCacheStore store = CreateStore(); + DateTimeOffset now = DateTimeOffset.UtcNow; + + for (int i = 0; i < 100; i++) + { + store.Set($"max-age-zero-{i}", new CacheEntry + { + StatusCode = (int)HttpStatusCode.OK, + Body = [1, 2, 3], + Headers = new Dictionary(), + ExpiresAt = now, + StoredAt = now + }); + } + + for (int i = 0; i < 100; i++) + { + store.TryGetValue($"max-age-zero-{i}", out _) + .Should().BeFalse("max-age=0 responses without a validator must not accumulate in memory"); + } } // ── Revalidation grace (real clock — IMemoryCache evicts on its own clock) diff --git a/Stampede.Http.Tests/Caching/UnsafeMethodInvalidationTests.cs b/Stampede.Http.Tests/Caching/UnsafeMethodInvalidationTests.cs index 81e4e80..9b3dce4 100644 --- a/Stampede.Http.Tests/Caching/UnsafeMethodInvalidationTests.cs +++ b/Stampede.Http.Tests/Caching/UnsafeMethodInvalidationTests.cs @@ -269,8 +269,99 @@ public async Task UnsafeMethod_OnlyInvalidatesTargetUri_OtherEntriesPreserved() usersCalls.Should().Be(1, "/users should still be served from cache"); } + // ── Invalidation issues no redundant reads ─────────────────────────────── + + [Fact] + public async Task Invalidation_DoesNotReadTheEntryBeforeRemovingIt() + { + // Probing with a read before removing only served to make the log/metric count confirmed + // deletions, at the cost of fetching the whole stored body over the network on a distributed + // store. Removal is idempotent, so the read must not happen. + CountingCacheStore counting = new(new MemoryCacheStore(new MemoryCache(new MemoryCacheOptions()))); + StubHandler stub = new(_ => new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent("body") }); + CachingMiddleware middleware = new(counting, _keyBuilder, _options) { InnerHandler = stub }; + HttpMessageInvoker invoker = new(middleware); + + _ = await invoker.SendAsync(new HttpRequestMessage(HttpMethod.Get, "https://api.test/probe"), CancellationToken.None); + + counting.Reset(); + + HttpRequestMessage post = new(HttpMethod.Post, "https://api.test/probe") + { + Content = new StringContent("payload") + }; + _ = await invoker.SendAsync(post, CancellationToken.None); + + counting.Removes.Should().Be(1, "the effective request URI must be invalidated"); + counting.Gets.Should().Be(0, "invalidation must not read the entry it is about to remove"); + } + + [Fact] + public async Task Invalidation_WithLocationAndContentLocation_IssuesOneRemovePerUri() + { + CountingCacheStore counting = new(new MemoryCacheStore(new MemoryCache(new MemoryCacheOptions()))); + StubHandler stub = new(req => + { + if (req.Method == HttpMethod.Get) + { + return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent("body") }; + } + + HttpResponseMessage created = new(HttpStatusCode.Created) + { + Content = new StringContent("created") + }; + created.Headers.Location = new Uri("https://api.test/items/42"); + created.Content.Headers.ContentLocation = new Uri("https://api.test/items/canonical"); + return created; + }); + CachingMiddleware middleware = new(counting, _keyBuilder, _options) { InnerHandler = stub }; + HttpMessageInvoker invoker = new(middleware); + + counting.Reset(); + + HttpRequestMessage post = new(HttpMethod.Post, "https://api.test/items") + { + Content = new StringContent("payload") + }; + _ = await invoker.SendAsync(post, CancellationToken.None); + + counting.Removes.Should().Be(3, "effective request URI, Location and Content-Location are each invalidated once"); + counting.Gets.Should().Be(0, "invalidation must not read the entries it is about to remove"); + } + // ── Helpers ────────────────────────────────────────────────────────────── + /// Counts the store operations issued by the middleware. + private sealed class CountingCacheStore(ICacheStore inner) : ICacheStore + { + private int _gets; + private int _removes; + + public int Gets => Volatile.Read(ref _gets); + public int Removes => Volatile.Read(ref _removes); + + public void Reset() + { + Volatile.Write(ref _gets, 0); + Volatile.Write(ref _removes, 0); + } + + public bool TryGetValue(string key, out CacheEntry? entry) + { + _ = Interlocked.Increment(ref _gets); + return inner.TryGetValue(key, out entry); + } + + public void Set(string key, CacheEntry entry) => inner.Set(key, entry); + + public void Remove(string key) + { + _ = Interlocked.Increment(ref _removes); + inner.Remove(key); + } + } + private CachingMiddleware BuildMiddleware(Func handler) { StubHandler stub = new(handler); diff --git a/Stampede.Http.Tests/Caching/VaryVariantCachingTests.cs b/Stampede.Http.Tests/Caching/VaryVariantCachingTests.cs index c7f1907..5437b86 100644 --- a/Stampede.Http.Tests/Caching/VaryVariantCachingTests.cs +++ b/Stampede.Http.Tests/Caching/VaryVariantCachingTests.cs @@ -205,6 +205,75 @@ public async Task Variants_WorkWithDistributedStore() (await enAgain.Content.ReadAsStringAsync(TestContext.Current.CancellationToken)).Should().Be("lang=en"); } + // ── Field-name normalization ───────────────────────────────────────────── + + [Fact] + public async Task VaryFieldNameCasing_DoesNotAffectVariantKeying() + { + // Field names are normalized once at store time, so an origin that changes the casing of its Vary + // header between responses must still resolve to the same variant. + int callCount = 0; + CachingMiddleware middleware = BuildPipeline( + new MemoryCacheStore(new MemoryCache(new MemoryCacheOptions())), + req => + { + callCount++; + HttpResponseMessage r = VaryingByLanguage(req); + r.Headers.Vary.Clear(); + r.Headers.Vary.Add(callCount == 1 ? "Accept-Language" : "ACCEPT-LANGUAGE"); + return r; + }); + + HttpMessageInvoker invoker = new(middleware); + const string url = "https://api.test/vary/casing"; + + _ = await invoker.SendAsync(Req(url, "en"), TestContext.Current.CancellationToken); + HttpResponseMessage second = await invoker.SendAsync(Req(url, "en"), TestContext.Current.CancellationToken); + + callCount.Should().Be(1, "the second request must hit the stored variant regardless of Vary casing"); + (await second.Content.ReadAsStringAsync(TestContext.Current.CancellationToken)).Should().Be("lang=en"); + } + + [Fact] + public async Task MultipleVaryFields_OrderIndependent_ResolveToTheSameVariant() + { + int callCount = 0; + CachingMiddleware middleware = BuildPipeline( + new MemoryCacheStore(new MemoryCache(new MemoryCacheOptions())), + req => + { + callCount++; + HttpResponseMessage r = new(HttpStatusCode.OK) { Content = new StringContent("body") }; + + // The origin lists the same two fields in a different order on each response. + if (callCount == 1) + { + r.Headers.Vary.Add("Accept-Language"); + r.Headers.Vary.Add("Accept-Encoding"); + } + else + { + r.Headers.Vary.Add("Accept-Encoding"); + r.Headers.Vary.Add("Accept-Language"); + } + + return r; + }); + + HttpMessageInvoker invoker = new(middleware); + const string url = "https://api.test/vary/order"; + + HttpRequestMessage first = Req(url, "en"); + first.Headers.TryAddWithoutValidation("Accept-Encoding", "gzip"); + _ = await invoker.SendAsync(first, TestContext.Current.CancellationToken); + + HttpRequestMessage second = Req(url, "en"); + second.Headers.TryAddWithoutValidation("Accept-Encoding", "gzip"); + _ = await invoker.SendAsync(second, TestContext.Current.CancellationToken); + + callCount.Should().Be(1, "Vary field order must not change the variant key"); + } + private sealed class StubTransport(Func handler) : HttpMessageHandler { protected override Task SendAsync(HttpRequestMessage request, CancellationToken ct) diff --git a/Stampede.Http.Tests/Coalescing/BoundedBodyReadTests.cs b/Stampede.Http.Tests/Coalescing/BoundedBodyReadTests.cs new file mode 100644 index 0000000..77137b5 --- /dev/null +++ b/Stampede.Http.Tests/Coalescing/BoundedBodyReadTests.cs @@ -0,0 +1,162 @@ +using Stampede.Http.Coalescing; +using Stampede.Http.Options; +using FluentAssertions; +using System.Net; + +namespace Stampede.Http.Tests.Coalescing; + +/// +/// Verifies that MaxResponseBodyBytes is enforced while the body is being read rather than after it has +/// been fully materialised, so an oversized response never gets allocated in full. +/// +public sealed class BoundedBodyReadTests +{ + [Fact] + public async Task DeclaredContentLengthOverLimit_RejectedWithoutReadingTheBody() + { + TrackingContent content = new(new byte[1024], declareLength: true); + RequestCoalescer coalescer = new(new CoalescerOptions { MaxResponseBodyBytes = 10 }); + + Func act = () => coalescer.ExecuteAsync( + new RequestKey("GET", "https://api.test/big"), + () => Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) { Content = content })); + + await act.Should().ThrowAsync().WithMessage("*MaxResponseBodyBytes*"); + + content.BytesRead.Should().Be(0, + "a declared Content-Length over the limit must be rejected before any of the body is read"); + } + + [Fact] + public async Task ChunkedBodyOverLimit_AbandonedPartWayThrough() + { + // No Content-Length (chunked): the limit can only be enforced while streaming. The read must stop + // shortly after crossing the limit instead of buffering all 4 MB. + TrackingContent content = new(new byte[4 * 1024 * 1024], declareLength: false); + RequestCoalescer coalescer = new(new CoalescerOptions { MaxResponseBodyBytes = 1024 }); + + Func act = () => coalescer.ExecuteAsync( + new RequestKey("GET", "https://api.test/chunked"), + () => Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) { Content = content })); + + await act.Should().ThrowAsync().WithMessage("*MaxResponseBodyBytes*"); + + content.BytesRead.Should().BeLessThan(4 * 1024 * 1024, + "an oversized chunked body must be abandoned mid-stream, not buffered in full"); + } + + [Fact] + public async Task BodyWithinLimit_ReadInFull() + { + byte[] payload = [1, 2, 3, 4, 5]; + TrackingContent content = new(payload, declareLength: true); + RequestCoalescer coalescer = new(new CoalescerOptions { MaxResponseBodyBytes = 1024 }); + + HttpResponseMessage response = await coalescer.ExecuteAsync( + new RequestKey("GET", "https://api.test/small"), + () => Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) { Content = content }), + TestContext.Current.CancellationToken); + + byte[] received = await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + received.Should().Equal(payload); + } + + [Fact] + public async Task ChunkedBodyWithinLimit_ReadInFull() + { + byte[] payload = [.. Enumerable.Range(0, 5000).Select(i => (byte)(i % 256))]; + TrackingContent content = new(payload, declareLength: false); + RequestCoalescer coalescer = new(new CoalescerOptions { MaxResponseBodyBytes = 1024 * 1024 }); + + HttpResponseMessage response = await coalescer.ExecuteAsync( + new RequestKey("GET", "https://api.test/chunked-ok"), + () => Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) { Content = content }), + TestContext.Current.CancellationToken); + + byte[] received = await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + received.Should().Equal(payload, "a chunked body within the limit must be reassembled exactly"); + } + + /// + /// An that records how much of its payload was actually read and can hide its + /// length to emulate a chunked response. + /// + private sealed class TrackingContent : HttpContent + { + private readonly byte[] _payload; + private readonly bool _declareLength; + private int _bytesRead; + + public TrackingContent(byte[] payload, bool declareLength) + { + _payload = payload; + _declareLength = declareLength; + + if (declareLength) + { + Headers.ContentLength = payload.Length; + } + } + + public int BytesRead => Volatile.Read(ref _bytesRead); + + protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) => + CreateContentReadStreamAsync().ContinueWith(t => t.Result.CopyToAsync(stream)).Unwrap(); + + protected override Task CreateContentReadStreamAsync() => + Task.FromResult(new CountingStream(_payload, count => Interlocked.Add(ref _bytesRead, count))); + + protected override bool TryComputeLength(out long length) + { + length = _payload.Length; + return _declareLength; + } + + /// A read-only stream over the payload that reports how many bytes were consumed. + private sealed class CountingStream(byte[] payload, Action onRead) : Stream + { + private int _position; + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => payload.Length; + + public override long Position + { + get => _position; + set => throw new NotSupportedException(); + } + + public override int Read(byte[] buffer, int offset, int count) => + Read(buffer.AsSpan(offset, count)); + + public override int Read(Span buffer) + { + int remaining = payload.Length - _position; + if (remaining <= 0) + { + return 0; + } + + int toCopy = Math.Min(remaining, buffer.Length); + payload.AsSpan(_position, toCopy).CopyTo(buffer[..toCopy]); + _position += toCopy; + onRead(toCopy); + + return toCopy; + } + + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => + new(Read(buffer.Span)); + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + Task.FromResult(Read(buffer.AsSpan(offset, count))); + + public override void Flush() { } + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } + } +} diff --git a/Stampede.Http.Tests/Integration/BufferedBodyReuseTests.cs b/Stampede.Http.Tests/Integration/BufferedBodyReuseTests.cs new file mode 100644 index 0000000..e209226 --- /dev/null +++ b/Stampede.Http.Tests/Integration/BufferedBodyReuseTests.cs @@ -0,0 +1,113 @@ +using Stampede.Http.Caching; +using Stampede.Http.Coalescing; +using Stampede.Http.Handlers; +using Stampede.Http.Options; +using FluentAssertions; +using Microsoft.Extensions.Caching.Memory; +using System.Net; + +namespace Stampede.Http.Tests.Integration; + +/// +/// Verifies that a response body materialised by the coalescer is handed to the caching layer without being +/// copied out and rebuffered again — the double copy every coalesced caller used to pay on a cache miss. +/// +public sealed class BufferedBodyReuseTests +{ + private static (CachingMiddleware Middleware, ICacheStore Cache) BuildPipeline( + Func origin) + { + ICacheStore cache = new MemoryCacheStore(new MemoryCache(new MemoryCacheOptions())); + + StubTransport transport = new(origin); + CoalescingHandler coalescing = new(new RequestCoalescer(new CoalescerOptions())) { InnerHandler = transport }; + CachingMiddleware caching = new(cache, new DefaultCacheKeyBuilder(), + new CacheOptions { DefaultTtl = TimeSpan.FromMinutes(5) }) + { + InnerHandler = coalescing + }; + + return (caching, cache); + } + + [Fact] + public async Task CachedBody_SharesTheArrayBufferedByTheCoalescer() + { + byte[] payload = [.. Enumerable.Range(0, 256).Select(i => (byte)i)]; + + (CachingMiddleware middleware, ICacheStore cache) = BuildPipeline(_ => + new HttpResponseMessage(HttpStatusCode.OK) { Content = new ByteArrayContent(payload) }); + + HttpMessageInvoker invoker = new(middleware); + _ = await invoker.SendAsync(new HttpRequestMessage(HttpMethod.Get, "https://api.test/reuse"), CancellationToken.None); + + cache.TryGetValue("GET:https://api.test/reuse", out CacheEntry? entry).Should().BeTrue(); + entry!.Body.Should().Equal(payload, "the stored body must match what the origin sent"); + } + + [Fact] + public async Task CoalescedWaiters_AllReceiveIndependentReadableResponses() + { + // Sharing one buffer across callers must not make the responses interfere with each other. + byte[] payload = [.. Enumerable.Range(0, 1024).Select(i => (byte)(i % 256))]; + TaskCompletionSource gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + int originCalls = 0; + + (CachingMiddleware middleware, _) = BuildPipeline(request => + { + _ = Interlocked.Increment(ref originCalls); + gate.Task.GetAwaiter().GetResult(); + return new HttpResponseMessage(HttpStatusCode.OK) { Content = new ByteArrayContent(payload) }; + }); + + HttpMessageInvoker invoker = new(middleware); + + Task[] callers = [.. Enumerable.Range(0, 8).Select(_ => + Task.Run(() => invoker.SendAsync(new HttpRequestMessage(HttpMethod.Get, "https://api.test/waiters"), CancellationToken.None)))]; + + await Task.Delay(100, TestContext.Current.CancellationToken); + gate.SetResult(); + + HttpResponseMessage[] responses = await Task.WhenAll(callers); + + foreach (HttpResponseMessage response in responses) + { + byte[] body = await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + body.Should().Equal(payload, "every coalesced caller must be able to read the full body independently"); + } + + Volatile.Read(ref originCalls).Should().Be(1, "the callers should have shared a single origin call"); + } + + [Fact] + public async Task OversizedBody_NotCachedAndStillReadableByTheCaller() + { + byte[] payload = new byte[4096]; + Random.Shared.NextBytes(payload); + + ICacheStore cache = new MemoryCacheStore(new MemoryCache(new MemoryCacheOptions())); + StubTransport transport = new(_ => + new HttpResponseMessage(HttpStatusCode.OK) { Content = new ByteArrayContent(payload) }); + CachingMiddleware middleware = new(cache, new DefaultCacheKeyBuilder(), + new CacheOptions { DefaultTtl = TimeSpan.FromMinutes(5), MaxBodySizeBytes = 128 }) + { + InnerHandler = transport + }; + + HttpMessageInvoker invoker = new(middleware); + HttpResponseMessage response = await invoker.SendAsync( + new HttpRequestMessage(HttpMethod.Get, "https://api.test/oversized"), CancellationToken.None); + + cache.TryGetValue("GET:https://api.test/oversized", out _) + .Should().BeFalse("a body over MaxBodySizeBytes must not be stored"); + + byte[] body = await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + body.Should().Equal(payload, "skipping the cache must not cost the caller its response body"); + } + + private sealed class StubTransport(Func handler) : HttpMessageHandler + { + protected override Task SendAsync(HttpRequestMessage request, CancellationToken ct) => + Task.FromResult(handler(request)); + } +} diff --git a/Stampede.Http/Caching/BackgroundRevalidationCoordinator.cs b/Stampede.Http/Caching/BackgroundRevalidationCoordinator.cs new file mode 100644 index 0000000..d20cf39 --- /dev/null +++ b/Stampede.Http/Caching/BackgroundRevalidationCoordinator.cs @@ -0,0 +1,61 @@ +using System.Collections.Concurrent; + +namespace Stampede.Http.Caching; + +/// +/// Tracks which cache keys currently have a stale-while-revalidate background refresh in flight +/// (RFC 5861 §3), so a given key is refreshed at most once at a time. +/// +/// +/// +/// This state deliberately lives outside . IHttpClientFactory rotates +/// handler chains — every two minutes by default — and keeps expired chains alive while their handlers are +/// still in use, so several instances can serve the same named client +/// concurrently. A dictionary owned by the handler would therefore deduplicate only within one chain, letting +/// two chains revalidate the same key simultaneously: exactly the duplicated origin load that +/// stale-while-revalidate exists to avoid. Registering this type as a per-client singleton makes the +/// deduplication hold for the lifetime of the client. +/// +/// +internal sealed class BackgroundRevalidationCoordinator +{ + private readonly ConcurrentDictionary _inflight = new(StringComparer.Ordinal); + + /// + /// Runs as a fire-and-forget background refresh for , + /// unless a refresh for that key is already in flight. + /// + /// + /// The key is claimed before the work is started, never from inside it: a refresh that finished before its + /// own registration completed would otherwise leave the key claimed forever, permanently blocking further + /// background revalidation of that entry. + /// + /// The cache key being refreshed. + /// + /// The refresh to run. It is responsible for handling its own failures; any exception that escapes is + /// swallowed here so it cannot surface as an unobserved task exception. + /// + public void Schedule(string key, Func revalidate) + { + if (!_inflight.TryAdd(key, 0)) + { + return; + } + + _ = Task.Run(async () => + { + try + { + await revalidate().ConfigureAwait(false); + } + catch + { + // The caller logs its own failures; nothing observes this task. + } + finally + { + _ = _inflight.TryRemove(key, out _); + } + }); + } +} diff --git a/Stampede.Http/Caching/CachingMiddleware.cs b/Stampede.Http/Caching/CachingMiddleware.cs index c1f93b7..140f760 100644 --- a/Stampede.Http/Caching/CachingMiddleware.cs +++ b/Stampede.Http/Caching/CachingMiddleware.cs @@ -1,9 +1,9 @@ -using Stampede.Http.Metrics; +using Stampede.Http.Internal; +using Stampede.Http.Metrics; using Stampede.Http.Options; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; -using System.Collections.Concurrent; using System.Net; using System.Net.Http.Headers; using System.Text; @@ -14,6 +14,7 @@ internal sealed partial class CachingMiddleware(ICacheStore cache, ICacheKeyBuilder keyBuilder, IOptionsMonitor optionsMonitor, string clientName, + BackgroundRevalidationCoordinator backgroundRevalidations, StampedeHttpMetrics? metrics = null, ILogger? logger = null, TimeProvider? timeProvider = null) : DelegatingHandler @@ -22,16 +23,17 @@ internal sealed partial class CachingMiddleware(ICacheStore cache, private readonly ILogger logger = logger ?? NullLogger.Instance; private readonly TimeProvider _timeProvider = timeProvider ?? TimeProvider.System; - private readonly ConcurrentDictionary _backgroundRevalidations = new(StringComparer.Ordinal); private CacheOptions Options => optionsMonitor.Get(clientName); /// - /// Convenience constructor for testing — wraps a static options instance. + /// Convenience constructor for testing — wraps a static options instance and gives this handler its own + /// background-revalidation scope. /// internal CachingMiddleware(ICacheStore cache, ICacheKeyBuilder keyBuilder, CacheOptions options, StampedeHttpMetrics? metrics = null, ILogger? logger = null, TimeProvider? timeProvider = null) - : this(cache, keyBuilder, new StaticOptionsMonitor(options), string.Empty, metrics, logger, timeProvider) { } + : this(cache, keyBuilder, new StaticOptionsMonitor(options), string.Empty, + new BackgroundRevalidationCoordinator(), metrics, logger, timeProvider) { } /// /// Determines whether the specified HTTP request is eligible for caching based on its method, headers, and content. @@ -114,13 +116,18 @@ HttpStatusCode.Gone or /// /// The cache entry containing the status code, response body, and headers to be used for constructing the HTTP /// response. + /// + /// When , the stored header fields are replayed over an empty body. Used for HEAD, which + /// repeats the header fields the equivalent GET would have sent — including Content-Type and + /// Content-Length — but carries no content (RFC 9110 §9.3.2). + /// /// An instance of HttpResponseMessage populated with the status code, body, and headers from the provided cache /// entry. - private HttpResponseMessage CreateResponse(CacheEntry entry) + private HttpResponseMessage CreateResponse(CacheEntry entry, bool includeBody = true) { HttpResponseMessage response = new((HttpStatusCode)entry.StatusCode) { - Content = new ByteArrayContent(entry.Body) + Content = new ByteArrayContent(includeBody ? entry.Body : []) }; foreach (KeyValuePair header in entry.Headers) @@ -131,6 +138,14 @@ private HttpResponseMessage CreateResponse(CacheEntry entry) } } + if (!includeBody) + { + // RFC 9110 §9.3.2 — the HEAD response reports the content length the equivalent GET would have sent. + // Set it from the stored body rather than relying on a stored Content-Length header: HttpContentHeaders + // computes that value lazily and does not enumerate it, so it is often absent from the entry. + response.Content.Headers.ContentLength = entry.Body.Length; + } + // §5.1 — Age: elapsed seconds since the response was stored long ageSeconds = Math.Max(0L, (long)(_timeProvider.GetUtcNow() - entry.StoredAt).TotalSeconds); response.Headers.Age = new TimeSpan(ageSeconds * TimeSpan.TicksPerSecond); @@ -153,24 +168,49 @@ private async Task StoreAsync(string key, HttpRequestMessage request, HttpRespon return; } + long maxBodySizeBytes = Options.MaxBodySizeBytes; + + // Skip oversized responses before touching the body. Buffering one only to discard it would allocate + // the whole payload — and would also consume a live network stream that the caller still has to read. + if (response.Content.Headers.ContentLength is long declaredLength && declaredLength > maxBodySizeBytes) + { + LogBodyTooLarge(key, declaredLength, maxBodySizeBytes); + return; + } + // Capture Last-Modified before replacing Content, since ByteArrayContent has no content headers. DateTimeOffset? capturedLastModified = response.Content.Headers.LastModified; - // Capture all content headers before replacing Content so they survive the swap. - List>> contentHeaders = [.. response.Content.Headers]; + byte[] body; - byte[] body = await response.Content.ReadAsByteArrayAsync(ct).ConfigureAwait(false); + if (response.Content is BufferedByteArrayContent buffered) + { + // Already materialised by the coalescer (or an inner cache layer): reuse the array rather than + // copying it out and rebuffering into a second ByteArrayContent. Under a stampede every waiter + // reaches this path, so the saving is one full body copy per coalesced caller. + body = buffered.Buffer; + } + else + { + // Capture all content headers before replacing Content so they survive the swap. + List>> contentHeaders = [.. response.Content.Headers]; - response.Content = new ByteArrayContent(body); + body = await response.Content.ReadAsByteArrayAsync(ct).ConfigureAwait(false); - // Restore original content headers (Content-Type, Content-Encoding, etc.) - foreach (KeyValuePair> header in contentHeaders) - { - response.Content.Headers.TryAddWithoutValidation(header.Key, header.Value); + // Reading consumed the original stream, so hand the caller a replayable copy. + response.Content = new BufferedByteArrayContent(body); + + // Restore original content headers (Content-Type, Content-Encoding, etc.) + foreach (KeyValuePair> header in contentHeaders) + { + response.Content.Headers.TryAddWithoutValidation(header.Key, header.Value); + } } - if (body.Length > Options.MaxBodySizeBytes) + // Chunked responses carry no Content-Length, so the limit can only be enforced after the read. + if (body.Length > maxBodySizeBytes) { + LogBodyTooLarge(key, body.Length, maxBodySizeBytes); return; } @@ -235,8 +275,7 @@ private async ValueTask WriteEntryAsync(string primaryKey, CacheEntry entry, Can return; } - string variantKey = BuildVariantKey(primaryKey, entry.VaryFields, - field => entry.VaryValues.TryGetValue(field, out string[]? values) ? values : []); + string variantKey = BuildVariantKey(primaryKey, entry.VaryFields, entry.VaryValues); await cache.SetAsync(variantKey, entry, ct).ConfigureAwait(false); await cache.SetAsync(primaryKey, CreateVaryMarker(entry), ct).ConfigureAwait(false); @@ -262,44 +301,97 @@ private async ValueTask WriteEntryAsync(string primaryKey, CacheEntry entry, Can return null; } - string variantKey = BuildVariantKey(primaryKey, entry.VaryFields, - field => request.Headers.TryGetValues(field, out IEnumerable? values) ? [.. values] : []); + string variantKey = BuildVariantKey(primaryKey, entry.VaryFields, request); return await cache.GetAsync(variantKey, ct).ConfigureAwait(false); } /// - /// Builds a Vary secondary cache key by appending the request's normalized values for each Vary field to the - /// primary key (RFC 9111 §4.1). Field names are sorted case-insensitively and values are lower-cased so the - /// key agrees with the case-insensitive comparison performed by . + /// Builds a Vary secondary cache key from the values carries for each Vary + /// field (RFC 9111 §4.1). Used on the read path, where the field names come from the stored marker. /// - private static string BuildVariantKey(string primaryKey, string[] varyFields, Func getValues) + private static string BuildVariantKey(string primaryKey, string[] normalizedFields, HttpRequestMessage request) { - string[] fields = [.. varyFields]; - Array.Sort(fields, StringComparer.OrdinalIgnoreCase); + StringBuilder sb = StartVariantKey(primaryKey); - StringBuilder sb = new(primaryKey.Length + 32); - sb.Append(primaryKey); - - foreach (string field in fields) + foreach (string field in normalizedFields) { - sb.Append(VariantKeySeparator).Append(field.ToLowerInvariant()).Append('='); + sb.Append(VariantKeySeparator).Append(field).Append('='); - string[] values = getValues(field); - for (int i = 0; i < values.Length; i++) + if (request.Headers.TryGetValues(field, out IEnumerable? values)) { - if (i > 0) - { - sb.Append(','); - } + AppendValues(sb, values); + } + } + + return sb.ToString(); + } - sb.Append(values[i].ToLowerInvariant()); + /// + /// Builds a Vary secondary cache key from the request values captured when the entry was stored + /// (RFC 9111 §4.1). Used on the write path. + /// + private static string BuildVariantKey(string primaryKey, string[] normalizedFields, IReadOnlyDictionary varyValues) + { + StringBuilder sb = StartVariantKey(primaryKey); + + foreach (string field in normalizedFields) + { + sb.Append(VariantKeySeparator).Append(field).Append('='); + + if (varyValues.TryGetValue(field, out string[]? values)) + { + AppendValues(sb, values); } } return sb.ToString(); } + private static StringBuilder StartVariantKey(string primaryKey) + { + return new StringBuilder(primaryKey.Length + 32).Append(primaryKey); + } + + /// + /// Appends a comma-separated, lower-cased rendering of so the key agrees with the + /// case-insensitive comparison performed by . + /// + private static void AppendValues(StringBuilder sb, IEnumerable values) + { + bool first = true; + + foreach (string value in values) + { + if (!first) + { + sb.Append(','); + } + + AppendLowerInvariant(sb, value); + first = false; + } + } + + /// + /// Appends in lower case without allocating an intermediate string. Header values + /// are short, so the common case folds through the stack. + /// + private static void AppendLowerInvariant(StringBuilder sb, string value) + { + const int StackAllocThreshold = 256; + + if (value.Length > StackAllocThreshold) + { + _ = sb.Append(value.ToLowerInvariant()); + return; + } + + Span buffer = stackalloc char[value.Length]; + int written = MemoryExtensions.ToLowerInvariant(value.AsSpan(), buffer); + _ = sb.Append(buffer[..written]); + } + /// /// Creates a Vary marker for the given representation. The marker carries no body and mirrors the /// representation's expiry/stale/validator metadata only so its eviction deadline matches the variant it @@ -321,9 +413,32 @@ private static string BuildVariantKey(string primaryKey, string[] varyFields, Fu IsVaryMarker = true }; + /// + /// Extracts the Vary field names, normalized once here rather than on every lookup: lower-cased and + /// sorted, so the secondary key is deterministic regardless of the order or casing the origin used. + /// + /// + /// Field names are matched case-insensitively, so lower-casing loses nothing — and the variant key is + /// rebuilt on every cache read, which is where copying, sorting and lower-casing the names again would + /// otherwise be paid. Lower-cased names sort identically under ordinal and case-insensitive comparison. + /// private static string[] ExtractVaryFields(HttpResponseMessage response) { - return response.Headers.Vary.Count == 0 ? [] : [.. response.Headers.Vary]; + if (response.Headers.Vary.Count == 0) + { + return []; + } + + string[] fields = [.. response.Headers.Vary]; + + for (int i = 0; i < fields.Length; i++) + { + fields[i] = fields[i].ToLowerInvariant(); + } + + Array.Sort(fields, StringComparer.Ordinal); + + return fields; } private static bool IsImmutableEntry(CacheControlHeaderValue? cc) @@ -614,9 +729,7 @@ private async Task HandleHeadAsync(HttpRequestMessage reque { metrics?.RecordCacheHit(HttpMethod.Head); LogCacheHit(getKey); - HttpResponseMessage headHit = CreateResponse(entry); - headHit.Content = new ByteArrayContent([]); - return headHit; + return CreateResponse(entry, includeBody: false); } // Stale entry with a validator — conditional HEAD revalidation @@ -642,9 +755,7 @@ private async Task HandleHeadAsync(HttpRequestMessage reque CacheEntry refreshed = RefreshFromNotModified(entry, revalResponse); await WriteEntryAsync(getKey, refreshed, ct).ConfigureAwait(false); metrics?.RecordCacheHit(HttpMethod.Head); - HttpResponseMessage headRefreshed = CreateResponse(refreshed); - headRefreshed.Content = new ByteArrayContent([]); - return headRefreshed; + return CreateResponse(refreshed, includeBody: false); } return revalResponse; @@ -826,48 +937,49 @@ private bool CanServeStaleWhileRevalidate(CacheEntry entry) /// private void ScheduleBackgroundRevalidation(string key, CacheEntry entry, HttpRequestMessage originalRequest) { - _ = _backgroundRevalidations.GetOrAdd(key, k => - Task.Run(async () => + // Snapshot the request headers now: the caller's HttpRequestMessage is disposed once its response is + // returned, which can happen before the background task starts. + HttpRequestMessage bgRequest = new(originalRequest.Method, originalRequest.RequestUri); + foreach (KeyValuePair> header in originalRequest.Headers) + { + _ = bgRequest.Headers.TryAddWithoutValidation(header.Key, header.Value); + } + + if (entry.ETag is not null) + { + _ = bgRequest.Headers.Remove("If-None-Match"); + _ = bgRequest.Headers.TryAddWithoutValidation("If-None-Match", entry.ETag); + } + else if (entry.LastModified is DateTimeOffset lastModified) + { + bgRequest.Headers.IfModifiedSince = lastModified; + } + + backgroundRevalidations.Schedule(key, async () => + { + try { - try - { - HttpRequestMessage bgRequest = new(originalRequest.Method, originalRequest.RequestUri); - foreach (KeyValuePair> header in originalRequest.Headers) - { - _ = bgRequest.Headers.TryAddWithoutValidation(header.Key, header.Value); - } - - if (entry.ETag is not null) - { - _ = bgRequest.Headers.Remove("If-None-Match"); - _ = bgRequest.Headers.TryAddWithoutValidation("If-None-Match", entry.ETag); - } - else if (entry.LastModified is DateTimeOffset lastModified) - { - bgRequest.Headers.IfModifiedSince = lastModified; - } - - HttpResponseMessage response = await base.SendAsync(bgRequest, CancellationToken.None).ConfigureAwait(false); - - if (response.StatusCode == HttpStatusCode.NotModified) - { - CacheEntry refreshed = RefreshFromNotModified(entry, response); - await WriteEntryAsync(key, refreshed, CancellationToken.None).ConfigureAwait(false); - } - else if (IsResponseCacheable(response)) - { - await StoreAsync(key, bgRequest, response, CancellationToken.None).ConfigureAwait(false); - } - } - catch (Exception ex) + HttpResponseMessage response = await base.SendAsync(bgRequest, CancellationToken.None).ConfigureAwait(false); + + if (response.StatusCode == HttpStatusCode.NotModified) { - LogBackgroundRevalidationFailed(key, ex); + CacheEntry refreshed = RefreshFromNotModified(entry, response); + await WriteEntryAsync(key, refreshed, CancellationToken.None).ConfigureAwait(false); } - finally + else if (IsResponseCacheable(response)) { - _ = _backgroundRevalidations.TryRemove(key, out _); + await StoreAsync(key, bgRequest, response, CancellationToken.None).ConfigureAwait(false); } - })); + } + catch (Exception ex) + { + LogBackgroundRevalidationFailed(key, ex); + } + finally + { + bgRequest.Dispose(); + } + }); } /// @@ -908,24 +1020,12 @@ private string BuildGetKey(Uri? uri) private async ValueTask InvalidateForUnsafeMethod(HttpRequestMessage request, HttpResponseMessage response, CancellationToken ct) { // §4.4 MUST — effective request URI - string effectiveKey = BuildGetKey(request.RequestUri); - if (await cache.GetAsync(effectiveKey, ct).ConfigureAwait(false) is not null) - { - await cache.RemoveAsync(effectiveKey, ct).ConfigureAwait(false); - metrics?.RecordCacheInvalidation(); - LogCacheInvalidation(effectiveKey, request.Method.Method); - } + await InvalidateKeyAsync(BuildGetKey(request.RequestUri), request.Method, ct).ConfigureAwait(false); // §4.4 MAY — Location header if (response.Headers.Location is Uri location && location != request.RequestUri) { - string locationKey = BuildGetKey(location); - if (await cache.GetAsync(locationKey, ct).ConfigureAwait(false) is not null) - { - await cache.RemoveAsync(locationKey, ct).ConfigureAwait(false); - metrics?.RecordCacheInvalidation(); - LogCacheInvalidation(locationKey, request.Method.Method); - } + await InvalidateKeyAsync(BuildGetKey(location), request.Method, ct).ConfigureAwait(false); } // §4.4 MAY — Content-Location header @@ -933,16 +1033,26 @@ private async ValueTask InvalidateForUnsafeMethod(HttpRequestMessage request, Ht && contentLocation != request.RequestUri && contentLocation != response.Headers.Location) { - string contentLocationKey = BuildGetKey(contentLocation); - if (await cache.GetAsync(contentLocationKey, ct).ConfigureAwait(false) is not null) - { - await cache.RemoveAsync(contentLocationKey, ct).ConfigureAwait(false); - metrics?.RecordCacheInvalidation(); - LogCacheInvalidation(contentLocationKey, request.Method.Method); - } + await InvalidateKeyAsync(BuildGetKey(contentLocation), request.Method, ct).ConfigureAwait(false); } } + /// + /// Removes a single cache key as part of §4.4 invalidation. + /// + /// + /// The removal is issued unconditionally rather than probing with a read first: removal is idempotent, so + /// the read only served to make the log and metric count confirmed deletions — and against a distributed + /// store that meant fetching the whole stored body over the network just to decide whether to log, doubling + /// the round-trips of every successful unsafe request. The metric therefore counts invalidations issued. + /// + private async ValueTask InvalidateKeyAsync(string key, HttpMethod method, CancellationToken ct) + { + await cache.RemoveAsync(key, ct).ConfigureAwait(false); + metrics?.RecordCacheInvalidation(); + LogCacheInvalidation(key, method.Method); + } + /// /// Extracts all headers from the specified HTTP response, including both response and content headers. /// @@ -986,12 +1096,15 @@ private static Dictionary ExtractHeaders(HttpResponseMessage r [LoggerMessage(Level = LogLevel.Debug, Message = "Cache: storing response for {CacheKey}")] private partial void LogCacheStore(string cacheKey); + [LoggerMessage(Level = LogLevel.Debug, Message = "Cache: not storing {CacheKey}, body of {BodyBytes} bytes exceeds MaxBodySizeBytes ({MaxBodySizeBytes})")] + private partial void LogBodyTooLarge(string cacheKey, long bodyBytes, long maxBodySizeBytes); + [LoggerMessage(Level = LogLevel.Debug, Message = "Cache: serving stale-while-revalidate for {CacheKey}")] private partial void LogStaleWhileRevalidate(string cacheKey); [LoggerMessage(Level = LogLevel.Warning, Message = "Cache: background revalidation failed for {CacheKey}")] private partial void LogBackgroundRevalidationFailed(string cacheKey, Exception exception); - [LoggerMessage(Level = LogLevel.Information, Message = "Cache: invalidated {CacheKey} after successful {HttpMethod} request (RFC 9111 §4.4)")] + [LoggerMessage(Level = LogLevel.Debug, Message = "Cache: invalidating {CacheKey} after successful {HttpMethod} request (RFC 9111 §4.4)")] private partial void LogCacheInvalidation(string cacheKey, string httpMethod); } diff --git a/Stampede.Http/Caching/DistributedCacheStore.cs b/Stampede.Http/Caching/DistributedCacheStore.cs index 9c95a2b..b6e0736 100644 --- a/Stampede.Http/Caching/DistributedCacheStore.cs +++ b/Stampede.Http/Caching/DistributedCacheStore.cs @@ -56,7 +56,12 @@ public bool TryGetValue(string key, out CacheEntry? entry) /// public void Set(string key, CacheEntry entry) { - (byte[] bytes, DistributedCacheEntryOptions entryOptions) = Serialize(entry); + if (!TrySerialize(entry, out byte[] bytes, out DistributedCacheEntryOptions entryOptions)) + { + distributedCache.Remove(key); + return; + } + distributedCache.Set(key, bytes, entryOptions); } @@ -81,7 +86,12 @@ public void Remove(string key) /// public async ValueTask SetAsync(string key, CacheEntry entry, CancellationToken ct = default) { - (byte[] bytes, DistributedCacheEntryOptions entryOptions) = Serialize(entry); + if (!TrySerialize(entry, out byte[] bytes, out DistributedCacheEntryOptions entryOptions)) + { + await distributedCache.RemoveAsync(key, ct).ConfigureAwait(false); + return; + } + await distributedCache.SetAsync(key, bytes, entryOptions, ct).ConfigureAwait(false); } @@ -92,23 +102,31 @@ public async ValueTask RemoveAsync(string key, CancellationToken ct = default) } /// - /// Serializes to UTF-8 JSON using the source-generated context and - /// builds the with an AbsoluteExpiration - /// extended by the maximum stale window — plus the revalidation grace period when the entry - /// carries a validator — so entries remain available for stale-if-error / stale-while-revalidate - /// serving and conditional revalidation after . + /// Serializes to UTF-8 JSON using the source-generated context and builds the + /// with an AbsoluteExpiration extended by the maximum + /// stale window — plus the revalidation grace period when the entry carries a validator — so entries + /// remain available for stale-if-error / stale-while-revalidate serving and conditional revalidation + /// after . /// - private (byte[] Bytes, DistributedCacheEntryOptions Options) Serialize(CacheEntry entry) + /// + /// when the entry has no retention window left at all — it can never be served + /// nor revalidated, so writing it would only cost a round-trip and (on Redis) a negative TTL that deletes + /// the key straight away. Callers remove the key instead. + /// + private bool TrySerialize(CacheEntry entry, out byte[] bytes, out DistributedCacheEntryOptions entryOptions) { - byte[] bytes = JsonSerializer.SerializeToUtf8Bytes(entry, CacheEntryJsonContext.Default.CacheEntry); + TimeSpan retention = MemoryCacheStore.ComputeRetention(entry, options.RevalidationGraceSeconds); + + if (retention <= TimeSpan.Zero) + { + bytes = []; + entryOptions = new DistributedCacheEntryOptions(); + return false; + } - long staleWindowSeconds = Math.Max(entry.StaleIfErrorSeconds, entry.StaleWhileRevalidateSeconds); - long graceSeconds = MemoryCacheStore.HasValidator(entry) ? options.RevalidationGraceSeconds : 0; - long extensionSeconds = staleWindowSeconds + graceSeconds; - DateTimeOffset absoluteExpiration = extensionSeconds > 0 - ? entry.ExpiresAt + TimeSpan.FromSeconds(extensionSeconds) - : entry.ExpiresAt; + bytes = JsonSerializer.SerializeToUtf8Bytes(entry, CacheEntryJsonContext.Default.CacheEntry); + entryOptions = new DistributedCacheEntryOptions { AbsoluteExpiration = entry.StoredAt + retention }; - return (bytes, new DistributedCacheEntryOptions { AbsoluteExpiration = absoluteExpiration }); + return true; } } diff --git a/Stampede.Http/Caching/MemoryCacheStore.cs b/Stampede.Http/Caching/MemoryCacheStore.cs index d33e474..f1a6c73 100644 --- a/Stampede.Http/Caching/MemoryCacheStore.cs +++ b/Stampede.Http/Caching/MemoryCacheStore.cs @@ -33,30 +33,46 @@ public bool TryGetValue(string key, out CacheEntry? entry) /// public void Set(string key, CacheEntry entry) { + TimeSpan retention = ComputeRetention(entry, options.RevalidationGraceSeconds); + + if (retention <= TimeSpan.Zero) + { + // The entry has no freshness left, no stale window and no revalidation grace, so it can never be + // served nor revalidated. Retaining it would leak: IMemoryCache evicts by LRU only when a SizeLimit + // is configured (CacheOptions.MaxCacheSize), which is null by default — an entry stored without an + // expiration would then live for the lifetime of the process. Drop any previous representation at + // this key too, so a superseded response is never served. + memoryCache.Remove(key); + return; + } + using ICacheEntry cacheEntry = memoryCache.CreateEntry(key); cacheEntry.Value = entry; cacheEntry.Size = ComputeSize(entry); // Use a relative TTL so the eviction deadline is clock-agnostic (works with FakeTimeProvider in tests). - // The window is the freshness TTL plus the largest configured stale window so entries remain available - // for stale-if-error / stale-while-revalidate after they become stale, plus — when the entry carries a - // validator (ETag / Last-Modified) — the revalidation grace period, so a conditional If-None-Match / - // If-Modified-Since request is still possible after all serve-stale windows have elapsed. - // Truncate to whole seconds: FreshnessCalculator makes a separate GetUtcNow() call from StoreAsync, so - // ExpiresAt - StoredAt can be a few ticks positive for max-age=0 responses on the real clock. Treating - // sub-second residuals as zero keeps max-age=0 entries alive for conditional revalidation (ETag / LM). + cacheEntry.AbsoluteExpirationRelativeToNow = retention; + } + + /// + /// Computes how long an entry must be retained by the backing store: its freshness lifetime, plus the + /// largest configured stale window so it remains available for stale-if-error / stale-while-revalidate + /// after it becomes stale, plus — when it carries a validator (ETag / Last-Modified) — the + /// revalidation grace period, so a conditional If-None-Match / If-Modified-Since request is + /// still possible once all serve-stale windows have elapsed. + /// + /// + /// The retention window. Zero or negative means the entry can never be served nor revalidated. + /// A max-age=0 response with no validator lands here: FreshnessCalculator makes a separate + /// GetUtcNow() call from StoreAsync, so on the real clock the window is a few ticks rather + /// than exactly zero — small enough that the entry is evicted before it could ever be read back. + /// + internal static TimeSpan ComputeRetention(CacheEntry entry, long revalidationGraceSeconds) + { long staleWindowSeconds = Math.Max(entry.StaleIfErrorSeconds, entry.StaleWhileRevalidateSeconds); - long nominalTtlSeconds = (long)(entry.ExpiresAt - entry.StoredAt).TotalSeconds; - long graceSeconds = HasValidator(entry) ? options.RevalidationGraceSeconds : 0; - long evictionTtlSeconds = nominalTtlSeconds + staleWindowSeconds + graceSeconds; + long graceSeconds = HasValidator(entry) ? revalidationGraceSeconds : 0; - if (evictionTtlSeconds > 0) - { - cacheEntry.AbsoluteExpirationRelativeToNow = TimeSpan.FromSeconds(evictionTtlSeconds); - } - // evictionTtlSeconds <= 0 (e.g. max-age=0 with no stale window and no grace): omit expiration so the - // entry stays in memory and its ETag / Last-Modified can be used for conditional revalidation - // (LRU eviction only). + return (entry.ExpiresAt - entry.StoredAt) + TimeSpan.FromSeconds(staleWindowSeconds + graceSeconds); } /// diff --git a/Stampede.Http/Coalescing/CachedResponse.cs b/Stampede.Http/Coalescing/CachedResponse.cs index 8d55a57..ce9496e 100644 --- a/Stampede.Http/Coalescing/CachedResponse.cs +++ b/Stampede.Http/Coalescing/CachedResponse.cs @@ -1,4 +1,6 @@ -using System.Net; +using Stampede.Http.Internal; +using System.Buffers; +using System.Net; using System.Net.Http.Headers; namespace Stampede.Http.Coalescing; @@ -29,15 +31,9 @@ public static async Task FromResponseAsync( CancellationToken cancellationToken = default) { byte[] bodyBytes = response.Content is not null - ? await response.Content.ReadAsByteArrayAsync(cancellationToken).ConfigureAwait(false) + ? await ReadBoundedAsync(response.Content, maxBodyBytes, cancellationToken).ConfigureAwait(false) : []; - if (bodyBytes.Length > maxBodyBytes) - { - throw new InvalidOperationException( - $"Response body size ({bodyBytes.Length} bytes) exceeds the configured MaxResponseBodyBytes limit ({maxBodyBytes} bytes)."); - } - // RequestMessage is intentionally not cached. It is IDisposable, not thread-safe, // and sharing it across coalesced callers would cause subtle concurrency issues. return new CachedResponse( @@ -52,6 +48,82 @@ public static async Task FromResponseAsync( ); } + /// + /// Reads the response body, refusing to buffer more than . + /// + /// + /// The limit is enforced while reading rather than afterwards: checking the length of an + /// already-materialised array means a response far larger than the limit is fully allocated before it can + /// be rejected, which is exactly the case the limit exists to prevent. A declared Content-Length + /// rejects the response before a single byte is read; a chunked response is read incrementally and + /// abandoned as soon as it crosses the limit. + /// + /// The body exceeds . + private static async Task ReadBoundedAsync(HttpContent content, long maxBodyBytes, CancellationToken cancellationToken) + { + // Already materialised by an inner Stampede layer — reuse the array, there is no stream to bound. + if (content is BufferedByteArrayContent buffered) + { + return buffered.Buffer.Length > maxBodyBytes + ? throw BodyTooLarge(buffered.Buffer.Length, maxBodyBytes, exact: true) + : buffered.Buffer; + } + + long? declaredLength = content.Headers.ContentLength; + + if (declaredLength > maxBodyBytes) + { + throw BodyTooLarge(declaredLength.Value, maxBodyBytes, exact: true); + } + + if (declaredLength is not null) + { + // Length known and within the limit: let the framework allocate the array exactly once. + return await content.ReadAsByteArrayAsync(cancellationToken).ConfigureAwait(false); + } + + // Chunked / unknown length: copy incrementally so an oversized body is abandoned mid-stream. + Stream stream = await content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + + using MemoryStream sink = new(); + byte[] rented = ArrayPool.Shared.Rent(CopyBufferSize); + + try + { + long total = 0; + int read; + + while ((read = await stream.ReadAsync(rented, cancellationToken).ConfigureAwait(false)) > 0) + { + total += read; + + if (total > maxBodyBytes) + { + throw BodyTooLarge(total, maxBodyBytes, exact: false); + } + + sink.Write(rented, 0, read); + } + } + finally + { + ArrayPool.Shared.Return(rented); + } + + return sink.ToArray(); + } + + /// Matches the default copy buffer size used by . + private const int CopyBufferSize = 81920; + + private static InvalidOperationException BodyTooLarge(long observedBytes, long maxBodyBytes, bool exact) + { + string size = exact ? $"{observedBytes} bytes" : $"more than {observedBytes} bytes"; + + return new InvalidOperationException( + $"Response body size ({size}) exceeds the configured MaxResponseBodyBytes limit ({maxBodyBytes} bytes)."); + } + private static KeyValuePair>[] MaterializeHeaders( HttpHeaders headers) { @@ -93,7 +165,9 @@ public HttpResponseMessage ToHttpResponseMessage() if (BodyBytes.Length > 0 || ContentHeaders.Count > 0) { - ByteArrayContent content = new(BodyBytes); + // BufferedByteArrayContent, not ByteArrayContent: it lets the caching layer above reuse these + // bytes directly instead of copying the whole body out again on its way into the cache. + BufferedByteArrayContent content = new(BodyBytes); foreach (KeyValuePair> header in ContentHeaders) { _ = content.Headers.TryAddWithoutValidation(header.Key, header.Value); diff --git a/Stampede.Http/Extensions/HttpClientBuilderExtensions.cs b/Stampede.Http/Extensions/HttpClientBuilderExtensions.cs index e18f4a2..1c450b4 100644 --- a/Stampede.Http/Extensions/HttpClientBuilderExtensions.cs +++ b/Stampede.Http/Extensions/HttpClientBuilderExtensions.cs @@ -274,6 +274,12 @@ private static void AddHttpCache(IHttpClientBuilder builder, Action(clientName, (sp, _) => new MemoryCacheStore(sp.GetRequiredKeyedService(clientName), structuralOptions))); + // Per-client stale-while-revalidate deduplication. This must outlive the handler: IHttpClientFactory + // rotates handler chains, so state held by CachingMiddleware would let two live chains revalidate the + // same key at once. + builder.Services.TryAdd( + ServiceDescriptor.KeyedSingleton(clientName, (_, _) => new())); + // Backward compatibility: non-keyed resolution returns the first-registered client's services. builder.Services.TryAddSingleton(sp => sp.GetRequiredKeyedService(clientName)); @@ -286,6 +292,7 @@ private static void AddHttpCache(IHttpClientBuilder builder, Action(clientName), sp.GetRequiredService>(), clientName, + sp.GetRequiredKeyedService(clientName), sp.GetService(), sp.GetService()?.CreateLogger(), sp.GetService())); diff --git a/Stampede.Http/Internal/BufferedByteArrayContent.cs b/Stampede.Http/Internal/BufferedByteArrayContent.cs new file mode 100644 index 0000000..1649f71 --- /dev/null +++ b/Stampede.Http/Internal/BufferedByteArrayContent.cs @@ -0,0 +1,26 @@ +namespace Stampede.Http.Internal; + +/// +/// A that exposes the buffer it was built from, so a downstream layer in the +/// same pipeline can reuse the bytes instead of copying them out again. +/// +/// +/// +/// Both the coalescer and the cache hand responses to callers as fully buffered byte arrays. Without this +/// type the caching layer has no way to tell an already-materialised body from a live network stream, so it +/// must call ReadAsByteArrayAsync — which copies the whole payload again — and then rebuffer the +/// result into a fresh . On a coalesced miss that is two extra full-size +/// allocations per caller, and every waiter sharing one origin call pays them. +/// +/// +/// Sharing the array is safe: never mutates it, hands out copies from +/// ReadAsByteArrayAsync, and exposes it only through a read-only stream. This is the same sharing the +/// cache already relies on when it serves many responses from one stored CacheEntry.Body. +/// +/// +/// The response body. Must not be mutated after construction. +internal sealed class BufferedByteArrayContent(byte[] buffer) : ByteArrayContent(buffer) +{ + /// The body this content was built from. + public byte[] Buffer { get; } = buffer; +} diff --git a/Stampede.Http/Metrics/StampedeHttpMetrics.cs b/Stampede.Http/Metrics/StampedeHttpMetrics.cs index da8d2a6..8a0ea93 100644 --- a/Stampede.Http/Metrics/StampedeHttpMetrics.cs +++ b/Stampede.Http/Metrics/StampedeHttpMetrics.cs @@ -14,7 +14,7 @@ namespace Stampede.Http.Metrics; /// stampede_http.cache.revalidationsConditional revalidation requests (If-None-Match / If-Modified-Since). /// stampede_http.cache.stale_errors_servedStale responses served under stale-if-error (RFC 5861). /// stampede_http.cache.stale_while_revalidate_servedStale responses served immediately while a background revalidation was triggered (RFC 5861). -/// stampede_http.cache.invalidationsCache entries invalidated by successful unsafe method responses (RFC 9111 §4.4). +/// stampede_http.cache.invalidationsCache invalidations issued after successful unsafe method responses (RFC 9111 §4.4). /// stampede_http.coalescing.deduplicatedRequests that reused an in-flight coalesced response. /// stampede_http.coalescing.inflightCurrent number of in-flight coalesced requests at the origin. /// stampede_http.coalescing.timeoutsCoalesced waiters that timed out and fell back to independent execution. @@ -70,7 +70,7 @@ public StampedeHttpMetrics() _cacheInvalidations = _meter.CreateCounter( "stampede_http.cache.invalidations", unit: "entries", - description: "Number of cache entries invalidated by successful unsafe method responses (RFC 9111 §4.4)."); + description: "Number of cache invalidations issued after successful unsafe method responses (RFC 9111 §4.4). Removal is idempotent, so this counts keys targeted rather than entries confirmed present."); _coalescedDeduplicated = _meter.CreateCounter( "stampede_http.coalescing.deduplicated",