diff --git a/.github/workflows/build-test-publish.yml b/.github/workflows/build-test-publish.yml index d1ef781f..5bd465d5 100644 --- a/.github/workflows/build-test-publish.yml +++ b/.github/workflows/build-test-publish.yml @@ -8,7 +8,7 @@ on: workflow_dispatch: env: - RELEASE_VERSION: v3.16.0 + RELEASE_VERSION: v3.16.1 # Add workflow-level permissions permissions: diff --git a/.gitignore b/.gitignore index 8492fffb..3b90380e 100644 --- a/.gitignore +++ b/.gitignore @@ -412,3 +412,5 @@ dist/ **/.DS_Store .claude/projects .vscode/settings.json + +scrap/ diff --git a/BenchmarkTests/BenchmarkTests.csproj b/BenchmarkTests/BenchmarkTests.csproj index bcae0082..17e64277 100644 --- a/BenchmarkTests/BenchmarkTests.csproj +++ b/BenchmarkTests/BenchmarkTests.csproj @@ -11,7 +11,7 @@ - + diff --git a/NpgsqlRest/NpgsqlRest.csproj b/NpgsqlRest/NpgsqlRest.csproj index 533cf80c..ad48a8a2 100644 --- a/NpgsqlRest/NpgsqlRest.csproj +++ b/NpgsqlRest/NpgsqlRest.csproj @@ -29,10 +29,10 @@ true README.MD bin\$(Configuration)\$(AssemblyName).xml - 3.16.0 - 3.16.0 - 3.16.0 - 3.16.0 + 3.16.1 + 3.16.1 + 3.16.1 + 3.16.1 true @@ -42,7 +42,7 @@ - + diff --git a/NpgsqlRest/NpgsqlRestEndpoint.cs b/NpgsqlRest/NpgsqlRestEndpoint.cs index 9f70e955..cd90cd63 100644 --- a/NpgsqlRest/NpgsqlRestEndpoint.cs +++ b/NpgsqlRest/NpgsqlRestEndpoint.cs @@ -1,4 +1,5 @@ using System.Buffers; +using System.Collections.Concurrent; using System.Collections.Frozen; using System.Data; using System.IO.Pipelines; @@ -23,6 +24,36 @@ public partial class NpgsqlRestEndpoint( NpgsqlRestMetadataEntry entry, FrozenDictionary overloads) { + // Signals a non-value exit from inside a cache GetOrCreateAsync factory, so the result is never + // cached and concurrent waiters observe the same outcome. NoRow → the command returned no row + // (caller maps to 500); ResponseAlreadyWritten → PrepareCommand short-circuited via + // CommandCallbackAsync and already wrote the lead caller's response (caller just returns). + private enum CacheFactoryAbortReason { NoRow, ResponseAlreadyWritten } + + private sealed class CacheFactoryAbortException(CacheFactoryAbortReason reason) : Exception + { + public CacheFactoryAbortReason Reason { get; } = reason; + } + + // Per-cache-key execution gates for the streaming records/sets path. That path cannot use the + // value-shaped GetOrCreateAsync factory model because it streams rows to the client and disables + // caching mid-stream once a response exceeds MaxCacheableRows. The gate serializes concurrent + // requests for the same key: the lead executes and (when within limits) populates the cache, so + // the rest get a cache hit instead of re-executing; over-limit responses simply run one-at-a-time, + // which still caps concurrent DB executions per key at one. + // + // The outer map is keyed by the cache instance itself (reference identity, so distinct cache + // profiles never share a gate even if they reuse a key string) and an inner map holds one gate per + // key. SemaphoreSlim instances are never disposed — they hold no OS handle unless AvailableWaitHandle + // is touched, and their count is bounded by (cache instances × distinct keys), bounded by the schema. + private static readonly ConcurrentDictionary> _recordCacheGates + = new(ReferenceEqualityComparer.Instance); + + private static SemaphoreSlim GetRecordCacheGate(IRoutineCache cache, string key) => + _recordCacheGates + .GetOrAdd(cache, static _ => new(StringComparer.Ordinal)) + .GetOrAdd(key, static _ => new SemaphoreSlim(1, 1)); + public async Task InvokeAsync(HttpContext context, IServiceProvider serviceProvider, CancellationToken cancellationToken = default) { Routine routine = entry.Endpoint.Routine; @@ -40,6 +71,8 @@ public async Task InvokeAsync(HttpContext context, IServiceProvider serviceProvi string? commandText = null; bool shouldDispose = true; bool shouldCommit = true; + // Held (when non-null) for the streaming records/sets cache path; released in the outer finally. + SemaphoreSlim? recordCacheGate = null; IUploadHandler? uploadHandler = null; var shouldLog = Options.LogCommands && Logger != null; @@ -1721,28 +1754,9 @@ await Results.Problem( } } - // Check cache for passthrough proxy endpoints (those without proxy response parameters) + // Passthrough proxy endpoints (those without proxy response parameters) may be cached; + // the cache read/execute/coalesce is handled at the InvokeAsync call below. bool isPassthroughProxy = !endpoint.HasProxyResponseParameters; - if (isPassthroughProxy && - endpoint.Cached is true && - bypassCache is false && - resolvedCache is not null) - { - if (resolvedCache.Get(endpoint, cacheKeyString!, out var cachedProxyResponse)) - { - // Cache hit - return cached proxy response - if (cachedProxyResponse is Proxy.ProxyResponse cached) - { - if (shouldLog) - { - cmdLog?.AppendLine("/* proxy response from cache */"); - NpgsqlRestLogger.LogEndpoint(endpoint, cmdLog?.ToString() ?? "", commandText); - } - await Proxy.ProxyRequestHandler.WriteResponseAsync(context, cached, Options.ProxyOptions, cancellationToken); - return; - } - } - } // Build user context headers if UserContext is enabled Dictionary? userContextHeaders = null; @@ -1785,6 +1799,36 @@ bypassCache is false && } } + // Cacheable passthrough proxy: route the upstream call through GetOrCreateAsync so a burst + // of identical requests coalesces into a single upstream call (stampede protection), and + // the response is cached for subsequent hits. + bool proxyCacheable = isPassthroughProxy && endpoint.Cached is true && bypassCache is false && resolvedCache is not null; + if (proxyCacheable) + { + bool factoryRan = false; + var cachedProxy = await resolvedCache!.GetOrCreateAsync( + endpoint, + cacheKeyString!, + async ct => + { + factoryRan = true; + return await Proxy.ProxyRequestHandler.InvokeAsync(context, endpoint, body, command.Parameters, userContextHeaders, ct); + }, + cacheTtlOverride, + cancellationToken); + + if (shouldLog && factoryRan is false) + { + cmdLog?.AppendLine("/* proxy response from cache */"); + NpgsqlRestLogger.LogEndpoint(endpoint, cmdLog?.ToString() ?? "", commandText); + } + if (cachedProxy is Proxy.ProxyResponse cachedProxyResponse) + { + await Proxy.ProxyRequestHandler.WriteResponseAsync(context, cachedProxyResponse, Options.ProxyOptions, cancellationToken); + } + return; + } + var proxyResponse = await Proxy.ProxyRequestHandler.InvokeAsync(context, endpoint, body, command.Parameters, userContextHeaders, cancellationToken); if (endpoint.HasProxyResponseParameters) @@ -1794,12 +1838,7 @@ bypassCache is false && } else { - // No proxy parameters - return proxy response directly without invoking PostgreSQL - // Cache the proxy response if caching is enabled - if (endpoint.Cached is true && bypassCache is false && resolvedCache is not null) - { - resolvedCache.AddOrUpdate(endpoint, cacheKeyString!, proxyResponse, cacheTtlOverride); - } + // No proxy parameters and not cacheable - return proxy response directly without PostgreSQL. await Proxy.ProxyRequestHandler.WriteResponseAsync(context, proxyResponse, Options.ProxyOptions, cancellationToken); return; } @@ -2414,41 +2453,55 @@ await command.ExecuteNonQueryWithRetryAsync( object? valueResult; if (resolvedCache is not null && endpoint.Cached is true && bypassCache is false) { - if (resolvedCache.Get(endpoint, cacheKeyString!, out valueResult) is false) + // Route through GetOrCreateAsync so a burst of identical requests coalesces into a + // single connection-open + execution (stampede protection). The connection is opened + // by PrepareCommand *inside* the factory, so coalesced waiters never touch the DB. + bool factoryRan = false; + try { - if (await PrepareCommand(connection, command, commandText, context, endpoint, true, cancellationToken) is false) - { - return; - } - - await using var reader = await command.ExecuteReaderWithRetryAsync( - CommandBehavior.SequentialAccess, - endpoint.RetryStrategy, - cancellationToken, - errorCodePolicy: endpoint.ErrorCodePolicy ?? Options.ErrorHandlingOptions.DefaultErrorCodePolicy); - if (shouldLog) - { - NpgsqlRestLogger.LogEndpoint(endpoint, cmdLog?.ToString() ?? "", command.CommandText); - } - if (await reader.ReadAsync(cancellationToken)) - { - valueResult = descriptor.IsBinary ? reader.GetFieldValue(0) : reader.GetValue(0) as string; - resolvedCache.AddOrUpdate(endpoint, cacheKeyString!, valueResult, cacheTtlOverride); - } - else + valueResult = await resolvedCache.GetOrCreateAsync( + endpoint, + cacheKeyString!, + async ct => + { + factoryRan = true; + if (await PrepareCommand(connection, command, commandText, context, endpoint, true, ct) is false) + { + throw new CacheFactoryAbortException(CacheFactoryAbortReason.ResponseAlreadyWritten); + } + + await using var reader = await command.ExecuteReaderWithRetryAsync( + CommandBehavior.SequentialAccess, + endpoint.RetryStrategy, + ct, + errorCodePolicy: endpoint.ErrorCodePolicy ?? Options.ErrorHandlingOptions.DefaultErrorCodePolicy); + if (shouldLog) + { + NpgsqlRestLogger.LogEndpoint(endpoint, cmdLog?.ToString() ?? "", command.CommandText); + } + if (await reader.ReadAsync(ct)) + { + return descriptor.IsBinary ? reader.GetFieldValue(0) : reader.GetValue(0) as string; + } + throw new CacheFactoryAbortException(CacheFactoryAbortReason.NoRow); + }, + cacheTtlOverride, + cancellationToken); + } + catch (CacheFactoryAbortException abort) + { + if (abort.Reason == CacheFactoryAbortReason.NoRow) { Logger?.CouldNotReadCommand(command.CommandText, context.Request.Method, context.Request.Path); context.Response.StatusCode = (int)HttpStatusCode.InternalServerError; - return; } + // ResponseAlreadyWritten: PrepareCommand already wrote the lead caller's response. + return; } - else + if (factoryRan is false && shouldLog) { - if (shouldLog) - { - cmdLog?.AppendLine("/* from cache */"); - NpgsqlRestLogger.LogEndpoint(endpoint, cmdLog?.ToString() ?? "", commandText); - } + cmdLog?.AppendLine("/* from cache */"); + NpgsqlRestLogger.LogEndpoint(endpoint, cmdLog?.ToString() ?? "", commandText); } } else @@ -2566,6 +2619,13 @@ Options.AuthenticationOptions.DefaultDataProtector is not null && if (canCacheRecordsAndSets) { + // Stampede protection for the streaming set path: serialize concurrent requests + // for this key. The lead executes and (within MaxCacheableRows) populates the cache + // below, so the waiters that acquire the gate next fall straight into this cache-hit + // branch instead of re-executing. Released in the outer finally. + recordCacheGate = GetRecordCacheGate(resolvedCache!, cacheKeyString!); + await recordCacheGate.WaitAsync(cancellationToken); + if (resolvedCache!.Get(endpoint, cacheKeyString!, out var cachedResult)) { if (shouldLog) @@ -3227,6 +3287,10 @@ await Results.Problem( } finally { + // Release the records/sets stampede gate (if this request acquired it) so the next waiter + // for this key can proceed — by now the cache is either populated or the response was over-limit. + recordCacheGate?.Release(); + await writer.CompleteAsync(); // proxy_out: capture buffered function output and forward to upstream proxy diff --git a/NpgsqlRest/RoutineCache.cs b/NpgsqlRest/RoutineCache.cs index 674e4a3b..9eec91f1 100644 --- a/NpgsqlRest/RoutineCache.cs +++ b/NpgsqlRest/RoutineCache.cs @@ -14,6 +14,39 @@ public interface IRoutineCache /// void AddOrUpdate(RoutineEndpoint endpoint, string key, object? value, TimeSpan? overrideExpiration = null); bool Remove(string key); + + /// + /// Atomic get-or-compute. On a cache hit the cached value is returned and + /// is NOT invoked. On a miss the factory runs the underlying work (e.g. opening the connection, + /// executing SQL and serializing the response) and its result is stored before returning. + /// + /// Backends that override this provide stampede protection: a burst of concurrent calls with the + /// same coalesce into a single factory invocation; the remaining callers + /// await the in-flight result instead of each hitting the database. The default implementation + /// below performs a plain probe-then-compute with no coalescing, so pre-existing custom + /// implementations keep working unchanged (they simply gain no stampede protection). + /// + /// + /// The endpoint whose response is being cached (carries the default TTL). + /// The cache key; identical keys coalesce into one factory invocation. + /// Runs the underlying work when the cache is cold. Should respect the token. + /// When non-null, the per-request TTL to use instead of the endpoint default. + /// Cancels this caller's wait; see backend notes for coalescing semantics. + async ValueTask GetOrCreateAsync( + RoutineEndpoint endpoint, + string key, + Func> factory, + TimeSpan? overrideExpiration = null, + CancellationToken cancellationToken = default) + { + if (Get(endpoint, key, out var existing)) + { + return existing; + } + var result = await factory(cancellationToken); + AddOrUpdate(endpoint, key, result, overrideExpiration); + return result; + } } public static class CacheKeyHasher @@ -53,6 +86,11 @@ private class CacheEntry } private static readonly ConcurrentDictionary _cache = new(StringComparer.Ordinal); + // In-flight factory invocations, keyed by effective (post-hash) cache key. Used to coalesce a burst + // of identical concurrent requests into a single factory run (stampede protection). The value is a + // Lazy so that even if ConcurrentDictionary.GetOrAdd runs its value-factory more than once under + // contention, only the stored Lazy's task is ever started — guaranteeing a single execution. + private static readonly ConcurrentDictionary>> _inflight = new(StringComparer.Ordinal); private static Timer? _cleanupTimer; private static CacheOptions _options = new(); @@ -125,4 +163,66 @@ public bool Remove(string key) var effectiveKey = CacheKeyHasher.GetEffectiveKey(key, _options); return _cache.TryRemove(effectiveKey, out _); } + + public async ValueTask GetOrCreateAsync( + RoutineEndpoint endpoint, + string key, + Func> factory, + TimeSpan? overrideExpiration = null, + CancellationToken cancellationToken = default) + { + var effectiveKey = CacheKeyHasher.GetEffectiveKey(key, _options); + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + + if (Get(endpoint, key, out var cached)) + { + return cached; + } + + // Coalesce: the first caller for this key creates the in-flight task and runs the factory + // (with its own live connection/command and its own token); concurrent callers await it. + var lazy = _inflight.GetOrAdd( + effectiveKey, + _ => new Lazy>(() => RunFactoryAsync(endpoint, key, effectiveKey, factory, overrideExpiration, cancellationToken))); + + try + { + return await lazy.Value.WaitAsync(cancellationToken); + } + catch (OperationCanceledException) + { + if (cancellationToken.IsCancellationRequested) + { + // This caller itself was cancelled — propagate. + throw; + } + // The lead caller (whose token drove the shared run) was cancelled before this waiter. + // Its in-flight entry has already been removed; loop and retry — the value may now be + // cached, or this caller becomes the new lead and runs with its own live resources. + } + } + } + + private async Task RunFactoryAsync( + RoutineEndpoint endpoint, + string key, + string effectiveKey, + Func> factory, + TimeSpan? overrideExpiration, + CancellationToken cancellationToken) + { + try + { + var result = await factory(cancellationToken).ConfigureAwait(false); + AddOrUpdate(endpoint, key, result, overrideExpiration); + return result; + } + finally + { + // Always free the slot so a later miss (or a failed/cancelled run) can start fresh. + _inflight.TryRemove(effectiveKey, out _); + } + } } diff --git a/NpgsqlRestClient/HybridCacheWrapper.cs b/NpgsqlRestClient/HybridCacheWrapper.cs index 591094ab..302ef9da 100644 --- a/NpgsqlRestClient/HybridCacheWrapper.cs +++ b/NpgsqlRestClient/HybridCacheWrapper.cs @@ -82,6 +82,30 @@ public void AddOrUpdate(RoutineEndpoint endpoint, string key, object? value, Tim } } + public async ValueTask GetOrCreateAsync( + RoutineEndpoint endpoint, + string key, + Func> factory, + TimeSpan? overrideExpiration = null, + CancellationToken cancellationToken = default) + { + // Delegate straight to HybridCache.GetOrCreateAsync: N concurrent calls with the same key + // share ONE factory invocation (its built-in stampede protection). Values are stored as the + // serialized string form, matching AddOrUpdate above (binary payloads are not supported by the + // Hybrid backend, same as before). The factory's exceptions propagate without being cached. + var effectiveKey = GetEffectiveKey(key); + var expiry = overrideExpiration ?? endpoint.CacheExpiresIn; + var options = expiry.HasValue + ? new HybridCacheEntryOptions { Expiration = expiry.Value, LocalCacheExpiration = expiry.Value } + : new HybridCacheEntryOptions(); + + return await _cache.GetOrCreateAsync( + effectiveKey, + async ct => (await factory(ct))?.ToString(), + options, + cancellationToken: cancellationToken); + } + public bool Remove(string key) { try diff --git a/NpgsqlRestClient/NpgsqlRestClient.csproj b/NpgsqlRestClient/NpgsqlRestClient.csproj index ef6ac5c3..b2752e9b 100644 --- a/NpgsqlRestClient/NpgsqlRestClient.csproj +++ b/NpgsqlRestClient/NpgsqlRestClient.csproj @@ -10,7 +10,7 @@ true true full - 3.16.0 + 3.16.1 @@ -22,7 +22,7 @@ - + diff --git a/NpgsqlRestClient/RedisCache.cs b/NpgsqlRestClient/RedisCache.cs index 2ff8aa18..63c3000a 100644 --- a/NpgsqlRestClient/RedisCache.cs +++ b/NpgsqlRestClient/RedisCache.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using NpgsqlRest; using StackExchange.Redis; @@ -9,6 +10,10 @@ public class RedisCache : IRoutineCache, IDisposable private readonly IDatabase _db; private readonly ILogger? _logger; private readonly CacheOptions _cacheOptions; + // In-process in-flight factory invocations (single-instance coalescing). Cross-process + // coalescing across NpgsqlRest instances is intentionally out of scope; this only collapses + // a burst hitting THIS instance into one execution. See memory RoutineCache for the pattern. + private readonly ConcurrentDictionary>> _inflight = new(StringComparer.Ordinal); private bool _disposed; public RedisCache(string configuration, ILogger? logger = null, CacheOptions? cacheOptions = null) @@ -106,6 +111,63 @@ public void AddOrUpdate(RoutineEndpoint endpoint, string key, object? value, Tim } } + public async ValueTask GetOrCreateAsync( + RoutineEndpoint endpoint, + string key, + Func> factory, + TimeSpan? overrideExpiration = null, + CancellationToken cancellationToken = default) + { + var effectiveKey = GetEffectiveKey(key); + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + + if (Get(endpoint, key, out var cached)) + { + return cached; + } + + var lazy = _inflight.GetOrAdd( + effectiveKey, + _ => new Lazy>(() => RunFactoryAsync(endpoint, key, effectiveKey, factory, overrideExpiration, cancellationToken))); + + try + { + return await lazy.Value.WaitAsync(cancellationToken); + } + catch (OperationCanceledException) + { + if (cancellationToken.IsCancellationRequested) + { + throw; + } + // Lead caller cancelled before this waiter; retry (value may be cached now, or this + // caller becomes the new lead with its own live resources). + } + } + } + + private async Task RunFactoryAsync( + RoutineEndpoint endpoint, + string key, + string effectiveKey, + Func> factory, + TimeSpan? overrideExpiration, + CancellationToken cancellationToken) + { + try + { + var result = await factory(cancellationToken).ConfigureAwait(false); + AddOrUpdate(endpoint, key, result, overrideExpiration); + return result; + } + finally + { + _inflight.TryRemove(effectiveKey, out _); + } + } + public bool Remove(string key) { if (_disposed) diff --git a/NpgsqlRestTests/NpgsqlRestTests.csproj b/NpgsqlRestTests/NpgsqlRestTests.csproj index cd723e71..44af181c 100644 --- a/NpgsqlRestTests/NpgsqlRestTests.csproj +++ b/NpgsqlRestTests/NpgsqlRestTests.csproj @@ -16,9 +16,9 @@ - + - + runtime; build; native; contentfiles; analyzers; buildtransitive diff --git a/NpgsqlRestTests/RoutineCacheTests/CacheStampedeTests.cs b/NpgsqlRestTests/RoutineCacheTests/CacheStampedeTests.cs new file mode 100644 index 00000000..0d2bc5f0 --- /dev/null +++ b/NpgsqlRestTests/RoutineCacheTests/CacheStampedeTests.cs @@ -0,0 +1,220 @@ +namespace NpgsqlRestTests; + +public static partial class Database +{ + public static void CacheStampedeTests() + { + // Each test uses its own function/endpoint so the process-wide static memory cache from one + // test never warms another's key (xUnit does not guarantee method execution order). + // Every function records one row per actual execution and sleeps briefly, so a burst of + // concurrent identical requests must overlap inside the factory for coalescing to be observable. + script.Append(@" +create table cache_stampede_calls ( + id int generated always as identity primary key, + label text not null, + called_at timestamptz not null default clock_timestamp() +); + +create function cache_stampede_scalar() +returns text +language sql +as $$ + insert into cache_stampede_calls (label) values ('scalar'); + select pg_sleep(0.2); + select 'stampede-scalar-result'; +$$; +comment on function cache_stampede_scalar() is 'HTTP GET +cached'; + +create function cache_stampede_warm() +returns text +language sql +as $$ + insert into cache_stampede_calls (label) values ('warm'); + select pg_sleep(0.2); + select 'stampede-warm-result'; +$$; +comment on function cache_stampede_warm() is 'HTTP GET +cached'; + +create function cache_stampede_param(_k text) +returns text +language sql +as $$ + insert into cache_stampede_calls (label) values ('param:' || _k); + select pg_sleep(0.2); + select 'r:' || _k; +$$; +comment on function cache_stampede_param(text) is 'HTTP GET +cached _k'; + +-- Set-returning, within MaxCacheableRows (default 1000): a cold burst must coalesce to one execution. +create function cache_stampede_set() +returns table (n int, v text) +language sql +as $$ + insert into cache_stampede_calls (label) values ('set'); + select g.n, 'row-' || g.n::text from generate_series(1, 3) g(n), pg_sleep(0.2); +$$; +comment on function cache_stampede_set() is 'HTTP GET +cached'; + +-- Set-returning, ABOVE MaxCacheableRows (1001 > 1000): never cached, so the gate serializes but every +-- request still executes its own query. Verifies over-limit is handled correctly (no coalescing, no +-- corruption, no caching of a partial/over-limit result). +create function cache_stampede_set_big() +returns table (n int) +language sql +as $$ + insert into cache_stampede_calls (label) values ('setbig'); + select g.n from generate_series(1, 1001) g(n), pg_sleep(0.1); +$$; +comment on function cache_stampede_set_big() is 'HTTP GET +cached'; +"); + } +} + +[Collection("TestFixture")] +public class CacheStampedeTests(TestFixture test) +{ + private static async Task CountCallsAsync(string label) + { + await using var conn = Database.CreateConnection(); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = "select count(*) from cache_stampede_calls where label = $1"; + var p = cmd.CreateParameter(); + p.Value = label; + cmd.Parameters.Add(p); + return Convert.ToInt32(await cmd.ExecuteScalarAsync()); + } + + private static async Task CountCallsLikeAsync(string pattern) + { + await using var conn = Database.CreateConnection(); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = "select count(*) from cache_stampede_calls where label like $1"; + var p = cmd.CreateParameter(); + p.Value = pattern; + cmd.Parameters.Add(p); + return Convert.ToInt32(await cmd.ExecuteScalarAsync()); + } + + [Fact] + public async Task Test_Stampede_Coalesces_Concurrent_Cold_Requests_To_Single_Execution() + { + const int n = 50; + var tasks = Enumerable.Range(0, n) + .Select(_ => test.Client.GetAsync("/api/cache-stampede-scalar/")) + .ToArray(); + var responses = await Task.WhenAll(tasks); + + foreach (var r in responses) + { + r.StatusCode.Should().Be(HttpStatusCode.OK); + (await r.Content.ReadAsStringAsync()).Should().Be("stampede-scalar-result"); + r.Dispose(); + } + + (await CountCallsAsync("scalar")).Should().Be(1, + "a burst of identical cached requests must coalesce into exactly one DB execution"); + } + + [Fact] + public async Task Test_Stampede_Warm_Cache_Burst_Adds_No_Executions() + { + // Prime the cache with a single cold request. + using (var first = await test.Client.GetAsync("/api/cache-stampede-warm/")) + { + first.StatusCode.Should().Be(HttpStatusCode.OK); + (await first.Content.ReadAsStringAsync()).Should().Be("stampede-warm-result"); + } + (await CountCallsAsync("warm")).Should().Be(1, "the priming request executes exactly once"); + + // A subsequent burst is served entirely from cache - no factory runs at all. + var tasks = Enumerable.Range(0, 30) + .Select(_ => test.Client.GetAsync("/api/cache-stampede-warm/")) + .ToArray(); + var responses = await Task.WhenAll(tasks); + foreach (var r in responses) + { + r.StatusCode.Should().Be(HttpStatusCode.OK); + (await r.Content.ReadAsStringAsync()).Should().Be("stampede-warm-result"); + r.Dispose(); + } + + (await CountCallsAsync("warm")).Should().Be(1, "a warm-cache burst must add zero further executions"); + } + + [Fact] + public async Task Test_Stampede_Distinct_Keys_Do_Not_Coalesce() + { + const int perKey = 8; + var keys = new[] { "a", "b", "c", "d" }; + var tasks = keys + .SelectMany(k => Enumerable.Range(0, perKey) + .Select(_ => (key: k, task: test.Client.GetAsync($"/api/cache-stampede-param/?k={k}")))) + .ToArray(); + await Task.WhenAll(tasks.Select(t => t.task)); + + foreach (var (key, task) in tasks) + { + using var r = await task; + r.StatusCode.Should().Be(HttpStatusCode.OK); + (await r.Content.ReadAsStringAsync()).Should().Be($"r:{key}"); + } + + // Each distinct key coalesces independently -> exactly one execution per key. + (await CountCallsLikeAsync("param:%")).Should().Be(keys.Length, + "each distinct cache key coalesces on its own, so there is exactly one execution per key"); + } + + [Fact] + public async Task Test_Stampede_Set_Within_Limit_Coalesces_To_Single_Execution() + { + const int n = 50; + var tasks = Enumerable.Range(0, n) + .Select(_ => test.Client.GetAsync("/api/cache-stampede-set/")) + .ToArray(); + var responses = await Task.WhenAll(tasks); + + string? firstBody = null; + foreach (var r in responses) + { + r.StatusCode.Should().Be(HttpStatusCode.OK); + var body = await r.Content.ReadAsStringAsync(); + body.Should().StartWith("[").And.Contain("row-1").And.Contain("row-3"); + firstBody ??= body; + body.Should().Be(firstBody, "every coalesced caller must receive the identical cached set"); + r.Dispose(); + } + + (await CountCallsAsync("set")).Should().Be(1, + "a within-limit cached set must coalesce a concurrent burst into exactly one execution"); + } + + [Fact] + public async Task Test_Stampede_Set_Over_Limit_Executes_Per_Request_Without_Caching() + { + // 1001 rows exceeds the default MaxCacheableRows (1000), so nothing is cached. The gate + // serializes the requests but each still runs its own query and returns the full, correct set. + const int n = 4; + var tasks = Enumerable.Range(0, n) + .Select(_ => test.Client.GetAsync("/api/cache-stampede-set-big/")) + .ToArray(); + var responses = await Task.WhenAll(tasks); + + foreach (var r in responses) + { + r.StatusCode.Should().Be(HttpStatusCode.OK); + var body = await r.Content.ReadAsStringAsync(); + body.Should().StartWith("[").And.Contain("1001"); + r.Dispose(); + } + + (await CountCallsAsync("setbig")).Should().Be(n, + "an over-limit set is never cached, so each request executes its own query (serialized, not coalesced)"); + } +} diff --git a/changelog/v3.16.1.md b/changelog/v3.16.1.md new file mode 100644 index 00000000..303c9537 --- /dev/null +++ b/changelog/v3.16.1.md @@ -0,0 +1,57 @@ +# Changelog v3.16.1 (2026-06-01) + +## Version [3.16.1](https://github.com/NpgsqlRest/NpgsqlRest/tree/3.16.1) (2026-06-01) + +[Full Changelog](https://github.com/NpgsqlRest/NpgsqlRest/compare/3.16.0...3.16.1) + +Patch release that makes **cache stampede protection actually fire for cached routine responses**. The cache-options documentation has advertised stampede protection as a HybridCache feature, but the integration used `IRoutineCache` as a synchronous probe (`Get` / `AddOrUpdate`) that could not carry the SQL execution as the cache factory — so the protection never engaged. A burst of identical concurrent requests against a cold cache executed the underlying query *N* times, each taking a connection. In the worst case this exhausted Postgres' connection pool (`remaining connection slots are reserved for roles with the SUPERUSER attribute`), which combined with connection-retry backoff could pin the pool long enough to affect every app sharing the database. + +## What changed + +### `IRoutineCache` gains `GetOrCreateAsync` (additive) + +A new method routes the cold-cache work *through* the cache so concurrent callers for the same key coalesce into a single execution: + +```csharp +ValueTask GetOrCreateAsync( + RoutineEndpoint endpoint, + string key, + Func> factory, + TimeSpan? overrideExpiration = null, + CancellationToken cancellationToken = default); +``` + +It ships as a **default interface method** (plain probe → factory → store, no coalescing), so any pre-existing custom `IRoutineCache` implementation compiles and behaves exactly as before — it simply gains no stampede protection until it overrides the method. + +> Note: although this is a new public API surface (conventionally a minor bump), it is shipped as a patch because it fixes an advertised-but-broken feature and is fully backward compatible via the default implementation. + +### Stampede protection per backend + +- **Memory (`RoutineCache`)** and **Redis (`RedisCache`)** — coalesce concurrent factory invocations through an in-flight `ConcurrentDictionary>`. A burst collapses to one execution; the rest await the in-flight result. +- **HybridCache (`HybridCacheWrapper`)** — delegates straight to `HybridCache.GetOrCreateAsync`, so Microsoft's built-in stampede protection now genuinely engages. + +### Middleware paths + +- **Scalar** single-value and **passthrough proxy** responses (value-shaped) route through `GetOrCreateAsync`. The connection is opened *inside* the factory, so coalesced waiters never touch the database. The passthrough proxy case additionally coalesces identical upstream HTTP calls. +- **Records / sets** (the streaming path) use a **per-key execution gate** instead. This path streams rows to the client and disables caching mid-stream once a response exceeds `MaxCacheableRows` (default 1000), which does not fit the "compute one value, cache it, share it" factory model. The gate serializes concurrent requests for a key: the lead executes and (within the row limit) populates the cache, so the rest get a cache hit instead of re-executing. This caps concurrent DB executions per key at **one** in all cases — including over-limit responses, which serialize rather than run in parallel. + +## Effect + +A burst of *N* identical requests against a cold cache now results in **one** database execution (within-limit) or a single serialized execution at a time (over-limit), instead of *N* concurrent executions. The worst-case fan-out from one event is bounded by the number of distinct cache keys (bounded by the schema), not by the number of clients. + +## Test coverage (read this honestly) + +Automated coverage (`NpgsqlRestTests/RoutineCacheTests/CacheStampedeTests.cs`) runs against the **in-memory** backend with a live Postgres and asserts execution counts directly: + +- 50 concurrent cold scalar requests → exactly 1 execution; warm-cache burst → 0 further executions; 4 distinct keys → exactly 4 executions (one per key). +- 50 concurrent cold set requests (within limit) → exactly 1 execution; over-limit set (1001 rows) → one execution per request, never cached, all responses correct. + +The **HybridCache** path relies on Microsoft's own tested coalescing (`Microsoft.Extensions.Caching.Hybrid`) and the **Redis** path's coalescing is verified by inspection — neither is exercised by the test harness, which boots the core library with the default memory cache. Claims about those two backends are not backed by an automated test in this repo. + +## Known limitations + +- **Cross-process coalescing is out of scope.** Coalescing is in-process per NpgsqlRest instance; multiple instances each execute once. (HybridCache's Redis layer still shares the *cached* value across instances.) +- **Over-limit sets serialize, not coalesce.** Responses above `MaxCacheableRows` (default 1000) are never cached, so the per-key gate makes concurrent requests for such an endpoint run one-at-a-time rather than sharing a result. This is deliberate: it caps both concurrent DB executions *and* peak memory (only one large set renders per key at a time) — but it does reduce throughput for a cached endpoint that returns more than `MaxCacheableRows` rows under load. Since such an endpoint is never actually cached, the right fix when this matters is to raise `MaxCacheableRows` so the result caches and coalesces, or to drop the `cached` annotation (restoring fully concurrent, uncached execution). +- **The records/sets gate is held during the response stream.** Because the gate wraps streaming to the client (not just the DB read), a slow or stalled lead client can delay other clients requesting the *same* key until it finishes or its request cancels. Waiters honor their own cancellation token, so a waiter that gives up is never stuck. The scalar and proxy paths are unaffected — their coalescing slot covers only the upstream call, not the client write. +- **`CommandCallbackAsync` short-circuit under coalescing.** If a user-supplied `CommandCallbackAsync` short-circuits the response on a cached scalar endpoint, coalesced *waiters* (not the lead) may observe an empty response. This affects only that specific hook on a cached endpoint. +- **Cancellation.** The shared factory runs on the lead caller's token; if the lead cancels mid-flight, waiters retry (re-probe the cache, or one becomes the new lead). A waiter may rarely observe cancellation if the lead cancels at the exact moment of coalescing. This is a deliberate safety choice — the factory uses the lead's live connection, so fully detaching the shared work risks using a disposed connection. diff --git a/npm/package.json b/npm/package.json index dad0b104..daa179b8 100644 --- a/npm/package.json +++ b/npm/package.json @@ -1,6 +1,6 @@ { "name": "npgsqlrest", - "version": "3.16.0", + "version": "3.16.1", "description": "Automatic REST API for PostgreSQL Databases Client Build", "scripts": { "postinstall": "node postinstall.js", diff --git a/npm/postinstall.js b/npm/postinstall.js index 89e0ed9d..60c4eccb 100644 --- a/npm/postinstall.js +++ b/npm/postinstall.js @@ -5,7 +5,7 @@ const path = require("path"); const os = require("os"); const https = require("https"); -const downloadFrom = "https://github.com/NpgsqlRest/NpgsqlRest/releases/download/v3.16.0/"; +const downloadFrom = "https://github.com/NpgsqlRest/NpgsqlRest/releases/download/v3.16.1/"; // Download binary next to this script, not to ../.bin/ const binDir = path.join(__dirname, "bin");