From 5ef4df6a29b704262529ddde93d2fcfa99c0c928 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Francisco=20Ruiz=20Ad=C3=A1n?= Date: Sat, 25 Jul 2026 20:06:46 +0200 Subject: [PATCH] perf(core): halve allocations on the coalesced cache-miss path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Measured on a 256 KB body with 8 coalesced waiters (median of 40 runs): 8,948,560 B -> 4,748,376 B allocated per miss, 34.1x -> 18.1x the body size. The saving scales with the number of waiters, since each one used to pay its own full copy of the body. Body handling: - Reuse the buffer the coalescer already materialised instead of reading it out again and rebuffering into a second ByteArrayContent. The new internal BufferedByteArrayContent lets the caching layer tell an already-materialised body from a live network stream. - Enforce MaxResponseBodyBytes while reading rather than after: a declared Content-Length over the limit is rejected before a byte is read, and a chunked body is abandoned as soon as it crosses it. Previously an oversized response was fully allocated, then rejected. - Skip buffering entirely when the body exceeds MaxBodySizeBytes, so the caller keeps its untouched stream instead of paying for a copy that the cache then discards. Store behaviour: - Stop retaining entries with no freshness, no stale window and no validator: they can neither be served nor revalidated, and without MaxCacheSize there is no SizeLimit and therefore no LRU, so they lived for the lifetime of the process. The distributed store now skips the write too, rather than issuing one Redis deletes straight away via a negative TTL. - Drop the read that preceded every §4.4 invalidation. It only made the log and metric count confirmed deletions, at the cost of fetching the whole stored body over the network: 2 -> 1 round-trips in the common case, 6 -> 3 in the worst. stampede_http.cache.invalidations now counts invalidations issued; its description says so. - Normalize Vary field names once at store time (lower-cased, sorted) so each lookup only concatenates, instead of copying, sorting and lower-casing them again. Correctness, found along the way: - A HEAD served from cache dropped Content-Type and reported Content-Length: 0, against RFC 9110 section 9.3.2, because the content was replaced wholesale to empty the body. - stale-while-revalidate deduplication lived on CachingMiddleware, which IHttpClientFactory rotates every two minutes and can keep several of alive at once, so two chains could revalidate the same key together. Moved to a per-client BackgroundRevalidationCoordinator, which also claims the key before starting the work: a refresh that finished before its own registration landed used to leave the key claimed forever. Two MemoryCacheStoreExpirationTests expectations changed: they asserted that an expired entry with no validator stayed "for conditional revalidation", but there is nothing to revalidate with. The legitimate case they described is covered by RevalidationGraceSeconds. Note for distributed deployments: Vary markers written by earlier versions carry unnormalized field names, so the first lookup of each misses and rewrites it. Self-healing, not corruption. Co-Authored-By: Claude Opus 5 --- .../BackgroundRevalidationCoordinatorTests.cs | 107 ++++++ .../Caching/HeadMethodCachingTests.cs | 55 +++ .../MemoryCacheStoreExpirationTests.cs | 70 +++- .../Caching/UnsafeMethodInvalidationTests.cs | 91 +++++ .../Caching/VaryVariantCachingTests.cs | 69 ++++ .../Coalescing/BoundedBodyReadTests.cs | 162 +++++++++ .../Integration/BufferedBodyReuseTests.cs | 113 ++++++ .../BackgroundRevalidationCoordinator.cs | 61 ++++ Stampede.Http/Caching/CachingMiddleware.cs | 321 ++++++++++++------ .../Caching/DistributedCacheStore.cs | 50 ++- Stampede.Http/Caching/MemoryCacheStore.cs | 50 ++- Stampede.Http/Coalescing/CachedResponse.cs | 92 ++++- .../Extensions/HttpClientBuilderExtensions.cs | 7 + .../Internal/BufferedByteArrayContent.cs | 26 ++ Stampede.Http/Metrics/StampedeHttpMetrics.cs | 4 +- 15 files changed, 1122 insertions(+), 156 deletions(-) create mode 100644 Stampede.Http.Tests/Caching/BackgroundRevalidationCoordinatorTests.cs create mode 100644 Stampede.Http.Tests/Coalescing/BoundedBodyReadTests.cs create mode 100644 Stampede.Http.Tests/Integration/BufferedBodyReuseTests.cs create mode 100644 Stampede.Http/Caching/BackgroundRevalidationCoordinator.cs create mode 100644 Stampede.Http/Internal/BufferedByteArrayContent.cs 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",