fix(web): correct idempotent replay payload and serialize concurrent duplicates

src/BuildingBlocks/Web/Idempotency/IdempotencyEndpointFilter.cs

@@ -1,14 +1,15 @@
+using System.Collections.Concurrent;
 using System.Security.Cryptography;
 using System.Text;
 using System.Text.Json;
 using FSH.Framework.Caching;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Http;
 using Microsoft.Extensions.Caching.Distributed;
-using Microsoft.Extensions.Caching.Hybrid;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using StackExchange.Redis;
 
 namespace FSH.Framework.Web.Idempotency;
 
@@ -18,16 +19,24 @@ namespace FSH.Framework.Web.Idempotency;
 /// for subsequent requests with the same key.
 /// </summary>
 /// <remarks>
-/// Uses <see cref="IDistributedCache"/> directly for the probe read (bypassing
-/// <see cref="HybridCache"/>'s factory-mandatory API) and <see cref="HybridCache.SetAsync"/>
-/// for the write path so replays benefit from L1 and the regular tag invalidation story.
-/// Using <c>HybridCache</c> with <c>DisableUnderlyingData</c> as a "get-only probe" is a
-/// known anti-pattern tracked at dotnet/aspnetcore#57191.
+/// Uses <see cref="IDistributedCache"/> for both the probe read and the write, on the same raw key
+/// and serializer, so the two are symmetric (a HybridCache write keys its L2 entries under its own
+/// scheme, which a raw-key probe never finds — replay then silently never engages).
+/// The handler result is executed into a buffer so the cached payload is the real wire body and
+/// status code (an <c>Ok&lt;T&gt;</c>/<c>Created&lt;T&gt;</c> wrapper would otherwise be serialized
+/// verbatim, and <c>Response.StatusCode</c> is still the default at filter time — the IResult sets
+/// it only when it executes). Concurrent duplicate keys are serialized by an atomic in-flight
+/// reservation (Redis <c>SET NX</c> when a multiplexer is registered, an in-process set otherwise).
 /// </remarks>
 public sealed class IdempotencyEndpointFilter : IEndpointFilter
 {
     private static readonly JsonSerializerOptions JsonOpts = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase };
 
+    // In-process reservation used when no Redis multiplexer is registered. Single-instance only —
+    // a multi-instance host in this stack already runs Redis (shared Data Protection key ring), so
+    // the Redis branch below covers every deployment where cross-instance duplicates are possible.
+    private static readonly ConcurrentDictionary<string, byte> InFlight = new(StringComparer.Ordinal);
+
     public async ValueTask<object?> InvokeAsync(EndpointFilterInvocationContext context, EndpointFilterDelegate next)
     {
         ArgumentNullException.ThrowIfNull(context);
@@ -49,70 +58,197 @@ public sealed class IdempotencyEndpointFilter : IEndpointFilter
         }
 
         var distributedCache = httpContext.RequestServices.GetRequiredService<IDistributedCache>();
-        var hybridCache = httpContext.RequestServices.GetRequiredService<HybridCache>();
         var logger = httpContext.RequestServices.GetRequiredService<ILogger<IdempotencyEndpointFilter>>();
 
         // Include tenant context in cache key for isolation
         var tenantId = httpContext.User.FindFirst("tenant")?.Value ?? "global";
         var cacheKey = CacheKeys.IdempotencyEntry(tenantId, idempotencyKey);
-        var tags = new[] { CacheKeys.Tags.Idempotency, CacheKeys.Tags.Tenant(tenantId) };
 
         // Probe-only read via IDistributedCache (real GetAsync, null on miss — unlike HybridCache's
         // factory). Bypasses L1: replays are rare vs first-calls, so L1 warmth has little value.
-        var cachedBytes = await distributedCache.GetAsync(cacheKey, httpContext.RequestAborted).ConfigureAwait(false);
-        if (cachedBytes is not null && cachedBytes.Length > 0)
+        var cached = await ProbeAsync(distributedCache, cacheKey, httpContext.RequestAborted).ConfigureAwait(false);
+        if (cached is not null)
+        {
+            return await ReplayAsync(httpContext, cached, idempotencyKey, logger).ConfigureAwait(false);
+        }
+
+        // Atomically reserve the key so concurrent duplicates don't both execute the handler.
+        var multiplexer = httpContext.RequestServices.GetService<IConnectionMultiplexer>();
+        var reservationKey = cacheKey + ":inflight";
+        if (!await TryReserveAsync(multiplexer, reservationKey, options.ReservationTtl, logger, idempotencyKey, httpContext.RequestAborted).ConfigureAwait(false))
+        {
+            // Another request with this key is in flight. It may have finished between the probe
+            // and the reservation — re-probe once, otherwise report the in-progress conflict.
+            var raced = await ProbeAsync(distributedCache, cacheKey, httpContext.RequestAborted).ConfigureAwait(false);
+            return raced is not null
+                ? await ReplayAsync(httpContext, raced, idempotencyKey, logger).ConfigureAwait(false)
+                : TypedResults.Conflict("A request with this Idempotency-Key is already being processed.");
+        }
+
+        try
         {
-            var cached = JsonSerializer.Deserialize<CachedIdempotentResponse>(cachedBytes, JsonOpts);
-            if (cached is not null)
+            var result = await next(context).ConfigureAwait(false);
+
+            // Execute the result into a buffer to capture the real wire body + status code, then
+            // serve that buffer to the client. Returning the IResult unexecuted would leave
+            // Response.StatusCode at its default and cache the wrapper object, not the wire body.
+            var (statusCode, contentType, body) = await ExecuteAndCaptureAsync(result, httpContext).ConfigureAwait(false);
+
+            httpContext.Response.StatusCode = statusCode;
+            if (contentType is not null)
             {
-                if (logger.IsEnabled(LogLevel.Debug))
-                {
-                    logger.LogDebug("Idempotent replay for key {KeyHash}", HashKey(idempotencyKey));
-                }
-                httpContext.Response.Headers["Idempotency-Replayed"] = "true";
-                httpContext.Response.StatusCode = cached.StatusCode;
-                if (cached.ContentType is not null)
-                {
-                    httpContext.Response.ContentType = cached.ContentType;
-                }
+                httpContext.Response.ContentType = contentType;
+            }
 
-                if (cached.Body.Length > 0)
+            if (body.Length > 0)
+            {
+                await httpContext.Response.Body.WriteAsync(body, httpContext.RequestAborted).ConfigureAwait(false);
+            }
+
+            // Write to the SAME store + key the probe reads. HybridCache.SetAsync keys its L2 entries
+            // under its own scheme, so a raw-key IDistributedCache probe never found them and replay
+            // silently never engaged. Idempotency entries are short-lived (TTL) and their tag-purge
+            // path was unused, so IDistributedCache alone — symmetric with the probe — is correct.
+            try
+            {
+                var responseToCache = new CachedIdempotentResponse
                 {
-                    await httpContext.Response.Body.WriteAsync(cached.Body, httpContext.RequestAborted).ConfigureAwait(false);
-                }
+                    StatusCode = statusCode,
+                    ContentType = contentType ?? "application/json",
+                    Body = body,
+                };
 
-                return null; // Response already written
+                var payload = JsonSerializer.SerializeToUtf8Bytes(responseToCache, JsonOpts);
+                await distributedCache.SetAsync(
+                    cacheKey,
+                    payload,
+                    new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = options.DefaultTtl },
+                    httpContext.RequestAborted).ConfigureAwait(false);
+            }
+            // Best-effort caching: idempotency replay is a convenience, not a correctness requirement
+            catch (Exception ex) when (ex is not OperationCanceledException)
+            {
+                logger.LogWarning(ex, "Failed to cache idempotent response for key {KeyHash}", HashKey(idempotencyKey));
             }
+
+            // Response already written to the body directly; return an empty result so the framework
+            // doesn't serialize a null return and append "null" after the captured payload.
+            return Results.Empty;
+        }
+        finally
+        {
+            await ReleaseReservationAsync(multiplexer, reservationKey, logger, idempotencyKey).ConfigureAwait(false);
+        }
+    }
+
+    private static async ValueTask<CachedIdempotentResponse?> ProbeAsync(
+        IDistributedCache cache, string cacheKey, CancellationToken ct)
+    {
+        var bytes = await cache.GetAsync(cacheKey, ct).ConfigureAwait(false);
+        return bytes is { Length: > 0 }
+            ? JsonSerializer.Deserialize<CachedIdempotentResponse>(bytes, JsonOpts)
+            : null;
+    }
+
+    private static async ValueTask<object?> ReplayAsync(
+        HttpContext httpContext, CachedIdempotentResponse cached, string idempotencyKey, ILogger logger)
+    {
+        if (logger.IsEnabled(LogLevel.Debug))
+        {
+            logger.LogDebug("Idempotent replay for key {KeyHash}", HashKey(idempotencyKey));
+        }
+
+        httpContext.Response.Headers["Idempotency-Replayed"] = "true";
+        httpContext.Response.StatusCode = cached.StatusCode;
+        if (cached.ContentType is not null)
+        {
+            httpContext.Response.ContentType = cached.ContentType;
         }
 
-        // Execute the handler
-        var result = await next(context).ConfigureAwait(false);
+        if (cached.Body.Length > 0)
+        {
+            await httpContext.Response.Body.WriteAsync(cached.Body, httpContext.RequestAborted).ConfigureAwait(false);
+        }
+
+        // Empty result (not null) so the framework doesn't append a serialized "null".
+        return Results.Empty;
+    }
 
-        // Cache the response through HybridCache so the tag invalidation path works for purges.
+    private static async Task<(int StatusCode, string? ContentType, byte[] Body)> ExecuteAndCaptureAsync(
+        object? result, HttpContext httpContext)
+    {
+        var originalBody = httpContext.Response.Body;
+        await using var buffer = new MemoryStream();
+        httpContext.Response.Body = buffer;
         try
         {
-            var body = result is not null ? JsonSerializer.SerializeToUtf8Bytes(result, JsonOpts) : [];
-            var responseToCache = new CachedIdempotentResponse
+            switch (result)
             {
-                StatusCode = httpContext.Response.StatusCode is > 0 and < 600 ? httpContext.Response.StatusCode : 200,
-                ContentType = "application/json",
-                Body = body
-            };
+                case null:
+                    break;
+                case IResult endpointResult:
+                    await endpointResult.ExecuteAsync(httpContext).ConfigureAwait(false);
+                    break;
+                default:
+                    // A non-IResult return is serialized as JSON by the framework — mirror that.
+                    await httpContext.Response.WriteAsJsonAsync(result, result.GetType(), options: null, contentType: null, httpContext.RequestAborted).ConfigureAwait(false);
+                    break;
+            }
+
+            var statusCode = httpContext.Response.StatusCode is > 0 and < 600
+                ? httpContext.Response.StatusCode
+                : StatusCodes.Status200OK;
+            return (statusCode, httpContext.Response.ContentType, buffer.ToArray());
+        }
+        finally
+        {
+            httpContext.Response.Body = originalBody;
+        }
+    }
 
-            var setOptions = new HybridCacheEntryOptions
+    private static async ValueTask<bool> TryReserveAsync(
+        IConnectionMultiplexer? multiplexer, string reservationKey, TimeSpan ttl, ILogger logger, string idempotencyKey, CancellationToken ct)
+    {
+        if (multiplexer is not null)
+        {
+            try
+            {
+                var db = multiplexer.GetDatabase();
+                return await db.StringSetAsync(reservationKey, "1", ttl, When.NotExists).ConfigureAwait(false);
+            }
+            // Fail open on a Redis blip: the reservation is a concurrency convenience, not a correctness
+            // requirement (the response cache still dedups later retries). Proceed rather than 500 the
+            // request, matching the best-effort stance the response write already takes.
+            catch (Exception ex) when (ex is not OperationCanceledException)
             {
-                Expiration = options.DefaultTtl,
-                LocalCacheExpiration = options.DefaultTtl < TimeSpan.FromMinutes(2) ? options.DefaultTtl : TimeSpan.FromMinutes(2),
-            };
-            await hybridCache.SetAsync(cacheKey, responseToCache, setOptions, tags, httpContext.RequestAborted).ConfigureAwait(false);
+                logger.LogWarning(ex, "Idempotency reservation failed for key {KeyHash}; proceeding without it", HashKey(idempotencyKey));
+                return true;
+            }
         }
-        // Best-effort caching: idempotency replay is a convenience, not a correctness requirement
-        catch (Exception ex) when (ex is not OperationCanceledException)
+
+        _ = ct;
+        return InFlight.TryAdd(reservationKey, 0);
+    }
+
+    private static async ValueTask ReleaseReservationAsync(IConnectionMultiplexer? multiplexer, string reservationKey, ILogger logger, string idempotencyKey)
+    {
+        if (multiplexer is not null)
         {
-            logger.LogWarning(ex, "Failed to cache idempotent response for key {KeyHash}", HashKey(idempotencyKey));
+            try
+            {
+                await multiplexer.GetDatabase().KeyDeleteAsync(reservationKey).ConfigureAwait(false);
+            }
+            // Best-effort release: a Redis fault here must not throw out of the finally. The short
+            // ReservationTtl expires the key anyway, so a missed delete self-heals in seconds.
+            catch (Exception ex) when (ex is not OperationCanceledException)
+            {
+                logger.LogWarning(ex, "Failed to release idempotency reservation for key {KeyHash}", HashKey(idempotencyKey));
+            }
+
+            return;
         }
 
-        return result;
+        InFlight.TryRemove(reservationKey, out _);
     }
 
     private static string HashKey(string key)

src/BuildingBlocks/Web/Idempotency/IdempotencyOptions.cs

@@ -15,6 +15,15 @@ public sealed class IdempotencyOptions
     /// </summary>
     public TimeSpan DefaultTtl { get; set; } = TimeSpan.FromHours(24);
 
+    /// <summary>
+    /// Time-to-live for the in-flight reservation that serializes concurrent duplicate keys.
+    /// Decoupled from <see cref="DefaultTtl"/>: it must only outlast the handler's execution, so a
+    /// crash between reserving and releasing frees the key in seconds instead of stranding it for the
+    /// full response TTL (every retry would 409 until it expired). Must exceed the longest expected
+    /// handler runtime — if it lapses mid-request a concurrent duplicate can slip through. Default: 1 minute.
+    /// </summary>
+    public TimeSpan ReservationTtl { get; set; } = TimeSpan.FromMinutes(1);
+
     /// <summary>
     /// Maximum allowed length for the idempotency key. Default: 128 characters.
     /// </summary>