From ca4abc28b226554468b3c29c2c86e4316f4704c1 Mon Sep 17 00:00:00 2001 From: feruzm Date: Fri, 21 Aug 2026 07:54:06 +0000 Subject: [PATCH 1/9] Internal SSR RPC cache for the web tier's server renders The web tier fetched the same Hive RPC reads (accounts, profiles, communities, per-tag feeds, posts) straight from public nodes from every renderer process, each at full upstream latency and with nothing shared. POST /private-api/ssr/rpc answers an allowlisted read method from one cache per host: the upstream result is serialized once (JsJson, lone surrogates included) and the bytes are written to every reader unchanged; concurrent misses for one key make one upstream call, the same single flight pattern as the chain balance fetch. Per-method TTLs follow how fast each read legitimately changes for a page. A lookup that outlives the budget answers 504 while the fill completes in the background; an RPC error answers 502; either way the consumer falls back to its own pool. Both routes are gated by a shared header secret and otherwise answer exactly like unknown routes (Routes.Fallback is now reusable), so nothing is learnable from outside and the parity catalog sees no new behavior. BytesCache is a bounded LRU of byte arrays with per-entry TTL; MemCache deep-clones on every access and has no size bound, which is wrong for whole feed payloads served to many readers. GET /private-api/ssr/stats exposes per-method hit, miss, coalesced, error and timeout counters behind the same header; no per-request logging. --- README.md | 5 + dotnet/EcencyApi.Tests/EcencyApi.Tests.csproj | 5 + dotnet/EcencyApi.Tests/SsrRpcTests.cs | 293 +++++++++++++++++ dotnet/EcencyApi/Config.cs | 26 ++ dotnet/EcencyApi/Handlers/Routes.cs | 46 +-- dotnet/EcencyApi/Handlers/SsrRpc.cs | 304 ++++++++++++++++++ dotnet/EcencyApi/Infrastructure/BytesCache.cs | 94 ++++++ .../EcencyApi/Infrastructure/HiveRpcClient.cs | 6 +- dotnet/docker-compose.yml | 7 + 9 files changed, 766 insertions(+), 20 deletions(-) create mode 100644 dotnet/EcencyApi.Tests/SsrRpcTests.cs create mode 100644 dotnet/EcencyApi/Handlers/SsrRpc.cs create mode 100644 dotnet/EcencyApi/Infrastructure/BytesCache.cs diff --git a/README.md b/README.md index b891cbdf..892cebdd 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,11 @@ docker run -it --rm -p 4000:4000 \ | `HELIUS_API_KEY` | optional Helius API key added as an extra Solana RPC fallback | | `ETH_RPC_URLS` / `BNB_RPC_URLS` / `SOL_RPC_URLS` / `BTC_ESPLORA_URLS` | optional comma-separated endpoint lists overriding the built-in chain provider pools | | `Logging__LogLevel__Default` | log level (default `Warning`; set `Information` for per-request logs) | +| `SSR_INTERNAL_SECRET` | shared header secret that switches on the internal SSR RPC cache routes (`/private-api/ssr/*`); unset = they answer like unknown routes | +| `SSR_CACHE_BYTES` | byte budget of the SSR RPC cache, LRU beyond it (default 512 MiB) | +| `SSR_RPC_BUDGET_MS` | wall-clock budget for one SSR RPC lookup before it answers 504 while the fill completes (default `1500`) | +| `SSR_RPC_NODE_TIMEOUT_MS` | per-node timeout of the SSR RPC cache's own client, one attempt per node (default `1200`) | +| `SSR_RPC_NODES` | comma-separated node pool for that client (default: the shared pool) | ## Swarm diff --git a/dotnet/EcencyApi.Tests/EcencyApi.Tests.csproj b/dotnet/EcencyApi.Tests/EcencyApi.Tests.csproj index 097b09b1..49f5bb96 100644 --- a/dotnet/EcencyApi.Tests/EcencyApi.Tests.csproj +++ b/dotnet/EcencyApi.Tests/EcencyApi.Tests.csproj @@ -17,6 +17,11 @@ + + + + + diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs new file mode 100644 index 00000000..495308bf --- /dev/null +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -0,0 +1,293 @@ +using System.Net; +using System.Text; +using System.Text.Json.Nodes; +using EcencyApi.Handlers; +using EcencyApi.Infrastructure; +using Microsoft.AspNetCore.Http; +using Xunit; + +namespace EcencyApi.Tests; + +/// +/// The SSR RPC cache against a loopback stub node: one upstream call per key +/// under concurrency, hits until the TTL runs out, the allowlist and header +/// gates answering like an unknown route, the budget turning a slow upstream +/// into a 504 while the fill still completes, and the byte budget evicting. +/// +[Collection("ssr-rpc")] +public class SsrRpcTests +{ + /// Loopback JSON-RPC node answering any method with a result that + /// names the method and the hit number, after an optional delay. + private sealed class RpcStub : IAsyncDisposable + { + private readonly HttpListener _listener = new(); + public string Url { get; } + public int Hits; + public int DelayMs; + public bool RpcError; + + public RpcStub() + { + var l = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0); + l.Start(); + var port = ((IPEndPoint)l.LocalEndpoint).Port; + l.Stop(); + Url = $"http://127.0.0.1:{port}/"; + _listener.Prefixes.Add(Url); + _listener.Start(); + _ = Loop(); + } + + private async Task Loop() + { + while (_listener.IsListening) + { + HttpListenerContext ctx; + try { ctx = await _listener.GetContextAsync(); } + catch { return; } + var n = Interlocked.Increment(ref Hits); + string reqBody; + using (var reader = new StreamReader(ctx.Request.InputStream)) + { + reqBody = await reader.ReadToEndAsync(); + } + var method = JsonNode.Parse(reqBody)?["params"]?[1]?.GetValue() ?? "?"; + if (DelayMs > 0) await Task.Delay(DelayMs); + var body = RpcError + ? "{\"jsonrpc\":\"2.0\",\"id\":1,\"error\":{\"message\":\"stub error\"}}" + : "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"method\":\"" + method + "\",\"n\":" + n + ",\"text\":\"caf\\u00e9 \\ud83d\"}}"; + var bytes = Encoding.UTF8.GetBytes(body); + ctx.Response.StatusCode = 200; + ctx.Response.ContentType = "application/json"; + ctx.Response.ContentLength64 = bytes.Length; + try { await ctx.Response.OutputStream.WriteAsync(bytes); ctx.Response.Close(); } catch { } + } + } + + public async ValueTask DisposeAsync() + { + _listener.Stop(); + _listener.Close(); + await Task.CompletedTask; + } + } + + private static readonly SsrRpc.MethodPolicy Post = SsrRpc.Allowlist["bridge.get_post"]; + private static readonly SsrRpc.MethodPolicy Props = SsrRpc.Allowlist["condenser_api.get_dynamic_global_properties"]; + + private static void Use(RpcStub stub, long cacheBytes = 1 << 20, int budgetMs = 1500) + { + SsrRpc.Client = new HiveRpcClient(new[] { stub.Url }, timeoutMs: 1000, failoverThreshold: 1); + SsrRpc.Cache = new BytesCache(cacheBytes); + SsrRpc.BudgetMs = budgetMs; + SsrRpc.ResetForTests(); + } + + private static JsonObject P(string author, string permlink) => + new() { ["author"] = author, ["permlink"] = permlink }; + + [Fact] + public async Task Concurrent_misses_for_one_key_make_one_upstream_call() + { + await using var stub = new RpcStub { DelayMs = 200 }; + Use(stub); + var tasks = Enumerable.Range(0, 12).Select(_ => SsrRpc.Resolve(Post, P("a", "b"))).ToArray(); + var results = await Task.WhenAll(tasks); + Assert.Equal(1, stub.Hits); + Assert.Single(results.Where(r => r.Outcome == SsrRpc.Outcome.Miss)); + Assert.Equal(11, results.Count(r => r.Outcome == SsrRpc.Outcome.Coalesced)); + var bodies = results.Select(r => Encoding.UTF8.GetString(r.Bytes)).Distinct().ToArray(); + Assert.Single(bodies); + Assert.Contains("\"method\":\"get_post\"", bodies[0]); + } + + [Fact] + public async Task Second_call_is_a_hit_with_the_same_bytes_and_params_order_does_not_matter() + { + await using var stub = new RpcStub(); + Use(stub); + var first = await SsrRpc.Resolve(Post, P("a", "b")); + var reordered = new JsonObject { ["permlink"] = "b", ["author"] = "a" }; + var second = await SsrRpc.Resolve(Post, reordered); + Assert.Equal(SsrRpc.Outcome.Miss, first.Outcome); + Assert.Equal(SsrRpc.Outcome.Hit, second.Outcome); + Assert.Equal(first.Bytes, second.Bytes); + Assert.Equal(1, stub.Hits); + // A different post is a different key. + var other = await SsrRpc.Resolve(Post, P("a", "c")); + Assert.Equal(SsrRpc.Outcome.Miss, other.Outcome); + Assert.Equal(2, stub.Hits); + } + + [Fact] + public async Task Bytes_are_the_upstream_result_serialized_once_lone_surrogate_included() + { + await using var stub = new RpcStub(); + Use(stub); + var r = await SsrRpc.Resolve(Post, P("a", "b")); + var text = Encoding.UTF8.GetString(r.Bytes); + // The result object itself, not the JSON-RPC envelope. + Assert.StartsWith("{\"method\":\"get_post\"", text); + Assert.DoesNotContain("jsonrpc", text); + // JsJson re-emits the lone surrogate as an escape instead of throwing. + Assert.Contains("\\ud83d", text); + } + + [Fact] + public async Task Ttl_expiry_goes_upstream_again() + { + await using var stub = new RpcStub(); + Use(stub); + var shortLived = Props with { TtlMs = 150 }; + Assert.Equal(SsrRpc.Outcome.Miss, (await SsrRpc.Resolve(shortLived, new JsonArray())).Outcome); + Assert.Equal(SsrRpc.Outcome.Hit, (await SsrRpc.Resolve(shortLived, new JsonArray())).Outcome); + await Task.Delay(300); + Assert.Equal(SsrRpc.Outcome.Miss, (await SsrRpc.Resolve(shortLived, new JsonArray())).Outcome); + Assert.Equal(2, stub.Hits); + } + + [Fact] + public async Task Budget_exceeded_answers_timeout_while_the_fill_still_lands_in_the_cache() + { + await using var stub = new RpcStub { DelayMs = 400 }; + Use(stub, budgetMs: 100); + var r = await SsrRpc.Resolve(Post, P("slow", "post")); + Assert.Equal(SsrRpc.Outcome.Timeout, r.Outcome); + await Task.Delay(600); + SsrRpc.BudgetMs = 1500; + var again = await SsrRpc.Resolve(Post, P("slow", "post")); + Assert.Equal(SsrRpc.Outcome.Hit, again.Outcome); + Assert.Equal(1, stub.Hits); + } + + [Fact] + public async Task Rpc_level_error_is_reported_not_cached() + { + await using var stub = new RpcStub { RpcError = true }; + Use(stub); + var r = await SsrRpc.Resolve(Post, P("a", "b")); + Assert.Equal(SsrRpc.Outcome.RpcError, r.Outcome); + Assert.Equal("stub error", r.Error); + var again = await SsrRpc.Resolve(Post, P("a", "b")); + Assert.Equal(SsrRpc.Outcome.RpcError, again.Outcome); + Assert.Equal(2, stub.Hits); + } + + [Fact] + public async Task Unreachable_node_is_reported_as_unavailable() + { + SsrRpc.Client = new HiveRpcClient(new[] { "http://127.0.0.1:9/" }, timeoutMs: 500, failoverThreshold: 1); + SsrRpc.Cache = new BytesCache(1 << 20); + SsrRpc.BudgetMs = 1500; + SsrRpc.ResetForTests(); + var r = await SsrRpc.Resolve(Post, P("a", "b")); + Assert.Equal(SsrRpc.Outcome.Unavailable, r.Outcome); + } + + [Fact] + public void Byte_budget_evicts_least_recently_used_and_refuses_oversize() + { + var cache = new BytesCache(100); + cache.Set("a", new byte[40], 60_000); + cache.Set("b", new byte[40], 60_000); + Assert.True(cache.TryGet("a", out _)); // a is now most recent + cache.Set("c", new byte[40], 60_000); // evicts b + Assert.True(cache.TryGet("a", out _)); + Assert.False(cache.TryGet("b", out _)); + Assert.True(cache.TryGet("c", out _)); + Assert.Equal(80, cache.Bytes); + cache.Set("huge", new byte[101], 60_000); + Assert.False(cache.TryGet("huge", out _)); + Assert.Equal(2, cache.Count); + } + + [Fact] + public async Task Expired_entry_is_dropped_on_read() + { + var cache = new BytesCache(1000); + cache.Set("k", new byte[10], 50); + Assert.True(cache.TryGet("k", out _)); + await Task.Delay(120); + Assert.False(cache.TryGet("k", out _)); + Assert.Equal(0, cache.Count); + Assert.Equal(0, cache.Bytes); + } + + [Fact] + public void Canonical_key_sorts_object_keys_at_every_level_and_keeps_array_order() + { + var a = JsonNode.Parse("{\"z\":1,\"a\":{\"y\":[2,{\"d\":1,\"c\":2}],\"b\":null}}"); + var b = JsonNode.Parse("{\"a\":{\"b\":null,\"y\":[2,{\"c\":2,\"d\":1}]},\"z\":1}"); + Assert.Equal(SsrRpc.CacheKey(Post, a), SsrRpc.CacheKey(Post, b)); + var c = JsonNode.Parse("{\"a\":{\"b\":null,\"y\":[{\"c\":2,\"d\":1},2]},\"z\":1}"); + Assert.NotEqual(SsrRpc.CacheKey(Post, a), SsrRpc.CacheKey(Post, c)); + } + + private static DefaultHttpContext Request(string method, string path, string? header, string? body = null) + { + var ctx = new DefaultHttpContext(); + ctx.Request.Method = method; + ctx.Request.Path = path; + if (header != null) ctx.Request.Headers[SsrRpc.HeaderName] = header; + if (body != null) + { + ctx.Request.ContentType = "application/json"; + var bytes = Encoding.UTF8.GetBytes(body); + ctx.Request.Body = new MemoryStream(bytes); + ctx.Request.ContentLength = bytes.Length; + } + ctx.Response.Body = new MemoryStream(); + return ctx; + } + + private static string ResponseText(HttpContext ctx) + { + ctx.Response.Body.Position = 0; + return new StreamReader(ctx.Response.Body).ReadToEnd(); + } + + [Fact] + public async Task Without_the_secret_configured_both_routes_answer_like_unknown_routes() + { + // SSR_INTERNAL_SECRET is not set in the test environment. + Assert.Null(Config.SsrInternalSecret); + var post = Request("POST", "/private-api/ssr/rpc", "anything", "{\"api\":\"bridge\",\"method\":\"get_post\"}"); + await SsrRpc.Rpc(post); + Assert.Equal(404, post.Response.StatusCode); + Assert.Contains("Cannot POST /private-api/ssr/rpc", ResponseText(post)); + + var get = Request("GET", "/private-api/ssr/stats", "anything"); + await SsrRpc.Stats(get); + Assert.Equal(200, get.Response.StatusCode); + Assert.Contains("text/html", get.Response.ContentType); + Assert.DoesNotContain("methods", ResponseText(get)); + } + + [Fact] + public void Authorized_requires_the_configured_secret_and_a_matching_header() + { + // With no secret configured nothing authorizes, header or not. + Assert.False(SsrRpc.Authorized(Request("POST", "/x", null))); + Assert.False(SsrRpc.Authorized(Request("POST", "/x", ""))); + Assert.False(SsrRpc.Authorized(Request("POST", "/x", "guess"))); + } + + [Fact] + public void Allowlist_is_read_only_and_names_every_method_the_consumer_routes() + { + foreach (var key in new[] + { + "bridge.get_ranked_posts", "bridge.get_account_posts", "bridge.get_post", "bridge.get_discussion", + "bridge.get_profile", "bridge.get_profiles", "bridge.get_community", "bridge.list_communities", + "condenser_api.get_accounts", "condenser_api.get_content", + "condenser_api.get_dynamic_global_properties", "condenser_api.get_trending_tags", + }) + { + Assert.True(SsrRpc.Allowlist.ContainsKey(key), key); + Assert.True(SsrRpc.Allowlist[key].TtlMs > 0, key); + } + Assert.False(SsrRpc.Allowlist.ContainsKey("condenser_api.broadcast_transaction")); + Assert.False(SsrRpc.Allowlist.ContainsKey("database_api.get_accounts")); + } +} diff --git a/dotnet/EcencyApi/Config.cs b/dotnet/EcencyApi/Config.cs index 0a7c55bf..29172c8d 100644 --- a/dotnet/EcencyApi/Config.cs +++ b/dotnet/EcencyApi/Config.cs @@ -30,5 +30,31 @@ public static class Config public static string CaptchaMode { get; } = (Env("CAPTCHA_MODE") ?? "hard").Trim().ToLowerInvariant(); + // ---- SSR RPC cache (Handlers/SsrRpc.cs) ---- + // Shared secret the web tier sends on every call. Unset = the routes are + // switched off and answer exactly like unknown routes. + public static string? SsrInternalSecret { get; } = NonEmpty(Env("SSR_INTERNAL_SECRET")); + + // Total bytes of cached responses kept in memory (LRU beyond that). + public static long SsrCacheBytes { get; } = + long.TryParse(Env("SSR_CACHE_BYTES"), out var b) && b >= 0 ? b : 512L * 1024 * 1024; + + // Wall-clock budget for one lookup. The web tier gives up on the proxy a + // little later and falls back to its own node pool, so this must stay + // under that; a lookup that outlives it still completes and fills the cache. + public static int SsrBudgetMs { get; } = + int.TryParse(Env("SSR_RPC_BUDGET_MS"), out var ms) && ms > 0 ? ms : 1500; + + // Per-node timeout for the cache's own RPC client (one attempt per node). + public static int SsrNodeTimeoutMs { get; } = + int.TryParse(Env("SSR_RPC_NODE_TIMEOUT_MS"), out var nt) && nt > 0 ? nt : 1200; + + // Optional node pool for the cache's RPC client, comma-separated; defaults + // to the shared pool. Lets a deployment put its own node first. + public static string[]? SsrRpcNodes { get; } = + NonEmpty(Env("SSR_RPC_NODES"))?.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); + + private static string? NonEmpty(string? value) => string.IsNullOrWhiteSpace(value) ? null : value.Trim(); + private static string? Env(string name) => Environment.GetEnvironmentVariable(name); } diff --git a/dotnet/EcencyApi/Handlers/Routes.cs b/dotnet/EcencyApi/Handlers/Routes.cs index e4d40a51..7d8068a0 100644 --- a/dotnet/EcencyApi/Handlers/Routes.cs +++ b/dotnet/EcencyApi/Handlers/Routes.cs @@ -168,6 +168,10 @@ public static void Map(WebApplication app) app.MapPost("/private-api/chats-update", PrivateApi.ChatsUpdate); app.MapPost("/private-api/channel-add", PrivateApi.ChannelAdd); + // ---- SSR RPC cache (internal, header-gated; see SsrRpc.cs) ---- + app.MapPost("/private-api/ssr/rpc", SsrRpc.Rpc); + app.MapGet("/private-api/ssr/stats", SsrRpc.Stats); + // ---- Health check for docker swarm ---- app.MapGet("/healthcheck.json", async ctx => { @@ -181,28 +185,34 @@ public static void Map(WebApplication app) // ---- Fallback ---- // GET/HEAD -> the template page (Express .get("*", fallbackHandler)). // Everything else -> Express's default finalhandler 404. - app.MapFallback(async ctx => + app.MapFallback(Fallback); + } + + /// + /// The unmatched-route response, also used by gated routes to answer exactly + /// like a route that does not exist. + /// + public static async Task Fallback(HttpContext ctx) + { + var method = ctx.Request.Method; + if (HttpMethods.IsGet(method) || HttpMethods.IsHead(method)) { - var method = ctx.Request.Method; - if (HttpMethods.IsGet(method) || HttpMethods.IsHead(method)) + ctx.Response.StatusCode = 200; + ctx.Response.ContentType = "text/html; charset=utf-8"; + if (!HttpMethods.IsHead(method)) { - ctx.Response.StatusCode = 200; - ctx.Response.ContentType = "text/html; charset=utf-8"; - if (!HttpMethods.IsHead(method)) - { - await ctx.Response.WriteAsync(TemplateHtml); - } - return; + await ctx.Response.WriteAsync(TemplateHtml); } + return; + } - var message = $"Cannot {method} {EscapeHtml(EncodeUrl(ctx.Request.Path.Value ?? "/"))}"; - var html = - "\n\n\n\n" + - "Error\n\n\n
" + message + "
\n\n\n"; - ctx.Response.StatusCode = 404; - ctx.Response.ContentType = "text/html; charset=utf-8"; - await ctx.Response.WriteAsync(html); - }); + var message = $"Cannot {method} {EscapeHtml(EncodeUrl(ctx.Request.Path.Value ?? "/"))}"; + var html = + "\n\n\n\n" + + "Error\n\n\n
" + message + "
\n\n\n"; + ctx.Response.StatusCode = 404; + ctx.Response.ContentType = "text/html; charset=utf-8"; + await ctx.Response.WriteAsync(html); } /// diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs new file mode 100644 index 00000000..73d15a39 --- /dev/null +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -0,0 +1,304 @@ +using System.Collections.Concurrent; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json.Nodes; +using EcencyApi.Infrastructure; + +namespace EcencyApi.Handlers; + +/// +/// Internal read-through cache for the Hive RPC reads the web tier makes while +/// rendering pages on the server. +/// +/// Every renderer process used to fetch the same accounts, profiles, +/// communities and per-tag feeds straight from public nodes, each at full +/// upstream latency and with nothing shared between processes. This route +/// answers those reads from one cache per host: the upstream `result` is +/// serialized once (JsJson, lone-surrogate safe) and the bytes are written to +/// every reader unchanged; concurrent misses for one key make one upstream +/// call (single-flight, the same pattern as the chain balance fetch). +/// +/// Boundaries, all deliberate: +/// - Allowlisted read methods only, with a TTL each. Anything else, and any +/// call without the shared internal header, answers exactly like a route +/// that does not exist (Routes.Fallback), so nothing is learnable from +/// outside and the parity harness sees no new behavior. +/// - The response is the raw `result`, so the consumer sees exactly what a +/// direct node call returns. No envelope, no reshaping. +/// - A lookup that outlives the budget answers 504 while the upstream call +/// completes in the background and fills the cache; an RPC-level error +/// answers 502. Either way the consumer falls back to its own node pool. +/// - No per-request logging (invariant 6); counters on /stats instead. +/// +public static class SsrRpc +{ + internal sealed record MethodPolicy(string Api, string Method, int TtlMs) + { + public string Key => $"{Api}.{Method}"; + } + + // TTLs follow how fast each read legitimately changes for a page render: + // dynamic props every block, a created feed within seconds, a post's + // votes/payout within tens of seconds, a profile or community rarely. + internal static readonly IReadOnlyDictionary Allowlist = new[] + { + new MethodPolicy("bridge", "get_ranked_posts", 15_000), + new MethodPolicy("bridge", "get_account_posts", 30_000), + new MethodPolicy("bridge", "get_post", 30_000), + new MethodPolicy("bridge", "get_discussion", 30_000), + new MethodPolicy("bridge", "get_profile", 60_000), + new MethodPolicy("bridge", "get_profiles", 60_000), + new MethodPolicy("bridge", "get_community", 300_000), + new MethodPolicy("bridge", "list_communities", 300_000), + new MethodPolicy("condenser_api", "get_accounts", 30_000), + new MethodPolicy("condenser_api", "get_content", 30_000), + new MethodPolicy("condenser_api", "get_dynamic_global_properties", 3_000), + new MethodPolicy("condenser_api", "get_trending_tags", 300_000), + }.ToDictionary(p => p.Key, p => p); + + internal const string HeaderName = "X-Ecency-Internal"; + + // Replaceable for tests (loopback stub nodes, a small cache budget). + internal static HiveRpcClient Client = new( + Config.SsrRpcNodes ?? HiveClients.DefaultNodes, + timeoutMs: Config.SsrNodeTimeoutMs, + failoverThreshold: 1); + + internal static BytesCache Cache = new(Config.SsrCacheBytes); + + internal static int BudgetMs = Config.SsrBudgetMs; + + private static readonly ConcurrentDictionary> InFlight = new(); + + internal sealed class Counter + { + public long Hit, Miss, Coalesced, Error, Timeout; + // Upstream latency EWMA for misses, milliseconds. + public double UpstreamMs; + private readonly object _lock = new(); + + public void RecordUpstream(double ms) + { + lock (_lock) UpstreamMs = UpstreamMs == 0 ? ms : UpstreamMs * 0.8 + ms * 0.2; + } + } + + private static readonly ConcurrentDictionary Counters = new(); + + internal static Counter CounterFor(string key) => Counters.GetOrAdd(key, _ => new Counter()); + + internal static void ResetForTests() + { + InFlight.Clear(); + Counters.Clear(); + } + + // ---- auth ---------------------------------------------------------------- + + internal static bool Authorized(HttpContext ctx) + { + var secret = Config.SsrInternalSecret; + if (secret is null) return false; + if (!ctx.Request.Headers.TryGetValue(HeaderName, out var values)) return false; + var presented = values.ToString(); + if (presented.Length == 0) return false; + return CryptographicOperations.FixedTimeEquals( + Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(secret)); + } + + // ---- key ----------------------------------------------------------------- + + /// + /// One key per distinct call: method plus params with object keys sorted at + /// every level, so two call sites that build the same params in a different + /// order share an entry. + /// + internal static string CacheKey(MethodPolicy policy, JsonNode? @params) => + policy.Key + ":" + JsJson.Stringify(Canonical(@params)); + + internal static JsonNode? Canonical(JsonNode? node) + { + switch (node) + { + case JsonObject obj: + var sorted = new JsonObject(); + foreach (var key in obj.Select(kv => kv.Key).OrderBy(k => k, StringComparer.Ordinal)) + { + sorted[key] = Canonical(obj[key]); + } + return sorted; + case JsonArray arr: + var copy = new JsonArray(); + foreach (var item in arr) copy.Add(Canonical(item)); + return copy; + default: + return node?.DeepClone(); + } + } + + // ---- core ---------------------------------------------------------------- + + internal enum Outcome { Hit, Miss, Coalesced, Timeout, RpcError, Unavailable } + + internal readonly record struct Resolution(Outcome Outcome, byte[] Bytes, string? Error); + + /// + /// Answer one allowlisted call from the cache, or from one shared upstream + /// call, within the budget. Pure of HTTP so it can be exercised directly. + /// + internal static async Task Resolve(MethodPolicy policy, JsonNode? @params) + { + var key = CacheKey(policy, @params); + var counter = CounterFor(policy.Key); + + if (Cache.TryGet(key, out var cached)) + { + Interlocked.Increment(ref counter.Hit); + return new Resolution(Outcome.Hit, cached, null); + } + + var coalesced = true; + if (!InFlight.TryGetValue(key, out var pending)) + { + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var winner = InFlight.GetOrAdd(key, tcs.Task); + if (ReferenceEquals(winner, tcs.Task)) + { + coalesced = false; + pending = tcs.Task; + _ = Fill(policy, @params, key, counter, tcs); + } + else + { + pending = winner; + } + } + + if (coalesced) Interlocked.Increment(ref counter.Coalesced); + else Interlocked.Increment(ref counter.Miss); + + var finished = await Task.WhenAny(pending, Task.Delay(BudgetMs)); + if (!ReferenceEquals(finished, pending)) + { + Interlocked.Increment(ref counter.Timeout); + return new Resolution(Outcome.Timeout, Array.Empty(), "budget exceeded"); + } + + try + { + var bytes = await pending; + return new Resolution(coalesced ? Outcome.Coalesced : Outcome.Miss, bytes, null); + } + catch (HiveRpcClient.RpcException e) + { + Interlocked.Increment(ref counter.Error); + return new Resolution(Outcome.RpcError, Array.Empty(), e.Message); + } + catch (Exception e) + { + Interlocked.Increment(ref counter.Error); + return new Resolution(Outcome.Unavailable, Array.Empty(), e.Message); + } + } + + // The one upstream call behind a key. Runs to completion even when every + // waiter has given up, so the cache still gets filled for the next reader. + private static async Task Fill(MethodPolicy policy, JsonNode? @params, string key, Counter counter, + TaskCompletionSource tcs) + { + var started = Environment.TickCount64; + try + { + var result = await Client.Call(policy.Api, policy.Method, @params ?? new JsonObject()); + var bytes = Encoding.UTF8.GetBytes(result is null ? "null" : JsJson.Stringify(result)); + counter.RecordUpstream(Environment.TickCount64 - started); + Cache.Set(key, bytes, policy.TtlMs); + tcs.TrySetResult(bytes); + } + catch (Exception e) + { + tcs.TrySetException(e); + _ = tcs.Task.Exception; // observe: every waiter may already be gone + } + finally + { + InFlight.TryRemove(new KeyValuePair>(key, tcs.Task)); + } + } + + // ---- handlers ------------------------------------------------------------ + + public static async Task Rpc(HttpContext ctx) + { + if (!Authorized(ctx)) + { + await Routes.Fallback(ctx); + return; + } + + var body = await ctx.ReadBody(); + var api = body.Str("api"); + var method = body.Str("method"); + if (api is null || method is null || !Allowlist.TryGetValue($"{api}.{method}", out var policy)) + { + await Routes.Fallback(ctx); + return; + } + + var resolution = await Resolve(policy, body.Field("params")); + ctx.Response.Headers["X-Ssr-Cache"] = resolution.Outcome.ToString().ToUpperInvariant(); + switch (resolution.Outcome) + { + case Outcome.Hit: + case Outcome.Miss: + case Outcome.Coalesced: + ctx.Response.StatusCode = 200; + ctx.Response.ContentType = "application/json; charset=utf-8"; + ctx.Response.ContentLength = resolution.Bytes.Length; + await ctx.Response.Body.WriteAsync(resolution.Bytes); + return; + case Outcome.Timeout: + await ctx.SendJson(504, new JsonObject { ["error"] = "Upstream Timeout" }); + return; + default: + await ctx.SendJson(502, new JsonObject { ["error"] = resolution.Error ?? "Upstream Error" }); + return; + } + } + + public static async Task Stats(HttpContext ctx) + { + if (!Authorized(ctx)) + { + await Routes.Fallback(ctx); + return; + } + + var methods = new JsonObject(); + foreach (var kv in Counters.OrderBy(kv => kv.Key, StringComparer.Ordinal)) + { + var c = kv.Value; + methods[kv.Key] = new JsonObject + { + ["hit"] = Interlocked.Read(ref c.Hit), + ["miss"] = Interlocked.Read(ref c.Miss), + ["coalesced"] = Interlocked.Read(ref c.Coalesced), + ["error"] = Interlocked.Read(ref c.Error), + ["timeout"] = Interlocked.Read(ref c.Timeout), + ["upstream_ms"] = Math.Round(c.UpstreamMs, 1), + }; + } + ctx.Response.Headers.CacheControl = "no-store"; + await ctx.SendJson(200, new JsonObject + { + ["cache"] = new JsonObject + { + ["bytes"] = Cache.Bytes, + ["count"] = Cache.Count, + ["budget"] = Cache.Budget, + }, + ["budget_ms"] = BudgetMs, + ["methods"] = methods, + }); + } +} diff --git a/dotnet/EcencyApi/Infrastructure/BytesCache.cs b/dotnet/EcencyApi/Infrastructure/BytesCache.cs new file mode 100644 index 00000000..7931071f --- /dev/null +++ b/dotnet/EcencyApi/Infrastructure/BytesCache.cs @@ -0,0 +1,94 @@ +namespace EcencyApi.Infrastructure; + +/// +/// Bounded in-process cache of serialized responses, keyed by string, with a +/// per-entry TTL and least-recently-used eviction under a total-bytes budget. +/// +/// Exists for the SSR RPC cache: the values are whole JSON payloads (feeds run +/// to hundreds of KB) that are served to many readers unchanged, so they are +/// stored once as UTF-8 bytes and written straight to the response. MemCache +/// is not used for these on purpose: it deep-clones a JsonNode on every set and +/// every get, has no size bound, and would force a JsJson.Stringify per hit. +/// +/// Thread-safe via one lock; every operation is O(1) apart from eviction, +/// which removes as many tail entries as the budget requires. +/// +public sealed class BytesCache +{ + private sealed class Entry + { + public required byte[] Bytes; + public required long ExpiresAtMs; + public required LinkedListNode Node; + } + + private readonly Dictionary _map = new(); + // Head = least recently used, tail = most recently used. + private readonly LinkedList _lru = new(); + private readonly object _lock = new(); + private long _bytes; + + public BytesCache(long budgetBytes) + { + Budget = Math.Max(0, budgetBytes); + } + + public long Budget { get; } + + public long Bytes { get { lock (_lock) return _bytes; } } + + public int Count { get { lock (_lock) return _map.Count; } } + + private static long NowMs => Environment.TickCount64; + + /// Fresh entry or nothing; an expired entry is dropped on the way. + public bool TryGet(string key, out byte[] bytes) + { + lock (_lock) + { + if (_map.TryGetValue(key, out var entry)) + { + if (entry.ExpiresAtMs > NowMs) + { + _lru.Remove(entry.Node); + _lru.AddLast(entry.Node); + bytes = entry.Bytes; + return true; + } + RemoveLocked(key, entry); + } + } + bytes = Array.Empty(); + return false; + } + + /// + /// Store a value for . A value larger than the whole + /// budget is not stored (it would evict everything for one reader). + /// + public void Set(string key, byte[] bytes, int ttlMs) + { + if (ttlMs <= 0 || bytes.Length > Budget) return; + lock (_lock) + { + if (_map.TryGetValue(key, out var existing)) + { + RemoveLocked(key, existing); + } + var node = _lru.AddLast(key); + _map[key] = new Entry { Bytes = bytes, ExpiresAtMs = NowMs + ttlMs, Node = node }; + _bytes += bytes.Length; + while (_bytes > Budget && _lru.First is { } oldest && oldest != node) + { + RemoveLocked(oldest.Value, _map[oldest.Value]); + } + } + } + + private void RemoveLocked(string key, Entry entry) + { + _map.Remove(key); + _lru.Remove(entry.Node); + _bytes -= entry.Bytes.Length; + } +} diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index 151280fd..3b5cb015 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs @@ -335,7 +335,7 @@ public static class HiveClients // portfolio engine/chain layers came back empty for everyone. GetAccounts // also routes around such a node at runtime, but keeping them out of the pool // means correctness here does not depend on that fallback firing. - public static readonly HiveRpcClient Default = new(new[] + public static readonly string[] DefaultNodes = { "https://api.hive.blog", "https://api.deathwing.me", @@ -345,5 +345,7 @@ public static class HiveClients "https://hive-api.3speak.tv", "https://api.syncad.com", "https://api.c0ff33a.uk", - }); + }; + + public static readonly HiveRpcClient Default = new(DefaultNodes); } diff --git a/dotnet/docker-compose.yml b/dotnet/docker-compose.yml index 20b863e6..f2e61e02 100644 --- a/dotnet/docker-compose.yml +++ b/dotnet/docker-compose.yml @@ -28,6 +28,13 @@ services: - SOL_RPC_URLS - BTC_ESPLORA_URLS - Logging__LogLevel__Default + # SSR RPC cache (Handlers/SsrRpc.cs): the shared header secret switches the + # routes on; the rest tune the cache budget, lookup budget and node pool. + - SSR_INTERNAL_SECRET + - SSR_CACHE_BYTES + - SSR_RPC_BUDGET_MS + - SSR_RPC_NODE_TIMEOUT_MS + - SSR_RPC_NODES ports: - "4000:4000" # Bound container logs so they can never grow unchecked on the host From 4ae19ea4e7d4832bf593b98fb9b860b5ecfc9647 Mon Sep 17 00:00:00 2001 From: feruzm Date: Fri, 21 Aug 2026 08:02:24 +0000 Subject: [PATCH 2/9] review: recheck before filling, bound fills, purge expired first, require params A reader that loses the in-flight race to a fill that completes and leaves could win GetOrAdd with the value already cached; the fill now rechecks the cache before calling upstream. Fills in progress are bounded (SSR_RPC_MAX_FILLS, default 64) so a slow pool plus many distinct keys queues instead of piling up detached calls. Under byte pressure the cache drops expired entries anywhere in the list before evicting a live one, at most once per 30s per pass. params must be an array or an object and both the key and the upstream call use that same node. A whitespace-only SSR_RPC_NODES falls back to the shared pool, which is exposed read-only. The EWMA is read under its lock. --- dotnet/EcencyApi.Tests/SsrRpcTests.cs | 51 ++++++++++++++++++- dotnet/EcencyApi/Config.cs | 9 +++- dotnet/EcencyApi/Handlers/SsrRpc.cs | 45 +++++++++++++--- dotnet/EcencyApi/Infrastructure/BytesCache.cs | 21 ++++++++ .../EcencyApi/Infrastructure/HiveRpcClient.cs | 4 +- 5 files changed, 118 insertions(+), 12 deletions(-) diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 495308bf..f1d073b6 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -76,11 +76,12 @@ public async ValueTask DisposeAsync() private static readonly SsrRpc.MethodPolicy Post = SsrRpc.Allowlist["bridge.get_post"]; private static readonly SsrRpc.MethodPolicy Props = SsrRpc.Allowlist["condenser_api.get_dynamic_global_properties"]; - private static void Use(RpcStub stub, long cacheBytes = 1 << 20, int budgetMs = 1500) + private static void Use(RpcStub stub, long cacheBytes = 1 << 20, int budgetMs = 1500, int maxFills = 64) { SsrRpc.Client = new HiveRpcClient(new[] { stub.Url }, timeoutMs: 1000, failoverThreshold: 1); SsrRpc.Cache = new BytesCache(cacheBytes); SsrRpc.BudgetMs = budgetMs; + SsrRpc.FillGate = new SemaphoreSlim(maxFills, maxFills); SsrRpc.ResetForTests(); } @@ -185,6 +186,54 @@ public async Task Unreachable_node_is_reported_as_unavailable() Assert.Equal(SsrRpc.Outcome.Unavailable, r.Outcome); } + [Fact] + public async Task Fills_in_progress_are_bounded_so_distinct_keys_queue_instead_of_piling_up() + { + await using var stub = new RpcStub { DelayMs = 250 }; + Use(stub, maxFills: 1); + var started = Environment.TickCount64; + var a = SsrRpc.Resolve(Post, P("a", "1")); + var b = SsrRpc.Resolve(Post, P("a", "2")); + var results = await Task.WhenAll(a, b); + Assert.All(results, r => Assert.Equal(SsrRpc.Outcome.Miss, r.Outcome)); + // The second fill waited for the first: two delays back to back. + Assert.True(Environment.TickCount64 - started >= 450, "fills ran concurrently despite the bound"); + Assert.Equal(2, stub.Hits); + } + + [Fact] + public async Task Under_pressure_expired_entries_go_before_live_ones() + { + var cache = new BytesCache(100); + cache.Set("live-old", new byte[40], 60_000); + cache.Set("expired", new byte[40], 50); + await Task.Delay(120); + // Over budget now: the expired entry must be the one that goes, even + // though the live entry is the least recently used. + cache.Set("new", new byte[40], 60_000); + Assert.True(cache.TryGet("live-old", out _)); + Assert.False(cache.TryGet("expired", out _)); + Assert.True(cache.TryGet("new", out _)); + } + + [Fact] + public async Task Rpc_requires_structured_params_and_keys_the_call_on_them() + { + await using var stub = new RpcStub(); + Use(stub); + // Missing params answers like an unknown route and never reaches upstream. + var missing = Request("POST", "/private-api/ssr/rpc", null, "{\"api\":\"bridge\",\"method\":\"get_post\"}"); + await SsrRpc.Rpc(missing); + Assert.Equal(404, missing.Response.StatusCode); + Assert.Equal(0, stub.Hits); + // An array and an object are both legitimate shapes and distinct keys. + var arr = await SsrRpc.Resolve(Props, new JsonArray()); + var obj = await SsrRpc.Resolve(Props, new JsonObject()); + Assert.Equal(SsrRpc.Outcome.Miss, arr.Outcome); + Assert.Equal(SsrRpc.Outcome.Miss, obj.Outcome); + Assert.Equal(2, stub.Hits); + } + [Fact] public void Byte_budget_evicts_least_recently_used_and_refuses_oversize() { diff --git a/dotnet/EcencyApi/Config.cs b/dotnet/EcencyApi/Config.cs index 29172c8d..b0eccafa 100644 --- a/dotnet/EcencyApi/Config.cs +++ b/dotnet/EcencyApi/Config.cs @@ -52,7 +52,14 @@ public static class Config // Optional node pool for the cache's RPC client, comma-separated; defaults // to the shared pool. Lets a deployment put its own node first. public static string[]? SsrRpcNodes { get; } = - NonEmpty(Env("SSR_RPC_NODES"))?.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); + NonEmpty(Env("SSR_RPC_NODES"))?.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries) + is { Length: > 0 } nodes ? nodes : null; + + // Upper bound on upstream fills in progress at once. A fill outlives the + // request budget on purpose (it still lands in the cache), so without a + // bound a slow pool plus many distinct keys would pile up detached calls. + public static int SsrMaxConcurrentFills { get; } = + int.TryParse(Env("SSR_RPC_MAX_FILLS"), out var f) && f > 0 ? f : 64; private static string? NonEmpty(string? value) => string.IsNullOrWhiteSpace(value) ? null : value.Trim(); diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index 73d15a39..e4498be1 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -60,7 +60,7 @@ internal sealed record MethodPolicy(string Api, string Method, int TtlMs) // Replaceable for tests (loopback stub nodes, a small cache budget). internal static HiveRpcClient Client = new( - Config.SsrRpcNodes ?? HiveClients.DefaultNodes, + Config.SsrRpcNodes ?? HiveClients.DefaultNodes.ToArray(), timeoutMs: Config.SsrNodeTimeoutMs, failoverThreshold: 1); @@ -68,6 +68,10 @@ internal sealed record MethodPolicy(string Api, string Method, int TtlMs) internal static int BudgetMs = Config.SsrBudgetMs; + // Bounds detached fills: a fill outlives the request budget on purpose, so + // a slow pool plus many distinct keys must not pile up unbounded calls. + internal static SemaphoreSlim FillGate = new(Config.SsrMaxConcurrentFills, Config.SsrMaxConcurrentFills); + private static readonly ConcurrentDictionary> InFlight = new(); internal sealed class Counter @@ -81,6 +85,11 @@ public void RecordUpstream(double ms) { lock (_lock) UpstreamMs = UpstreamMs == 0 ? ms : UpstreamMs * 0.8 + ms * 0.2; } + + public double ReadUpstreamMs() + { + lock (_lock) return UpstreamMs; + } } private static readonly ConcurrentDictionary Counters = new(); @@ -145,8 +154,11 @@ internal enum Outcome { Hit, Miss, Coalesced, Timeout, RpcError, Unavailable } /// /// Answer one allowlisted call from the cache, or from one shared upstream /// call, within the budget. Pure of HTTP so it can be exercised directly. + /// `params` is the exact node the upstream call carries (an array for + /// condenser methods, an object for bridge), and the key is derived from + /// that same node. /// - internal static async Task Resolve(MethodPolicy policy, JsonNode? @params) + internal static async Task Resolve(MethodPolicy policy, JsonNode @params) { var key = CacheKey(policy, @params); var counter = CounterFor(policy.Key); @@ -203,13 +215,24 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode? @p // The one upstream call behind a key. Runs to completion even when every // waiter has given up, so the cache still gets filled for the next reader. - private static async Task Fill(MethodPolicy policy, JsonNode? @params, string key, Counter counter, + private static async Task Fill(MethodPolicy policy, JsonNode @params, string key, Counter counter, TaskCompletionSource tcs) { - var started = Environment.TickCount64; + var acquired = false; try { - var result = await Client.Call(policy.Api, policy.Method, @params ?? new JsonObject()); + // A reader can miss the cache, lose the race to another fill that + // then completes and leaves, and only now win GetOrAdd: the value is + // already cached, so serve it rather than calling upstream again. + if (Cache.TryGet(key, out var cached)) + { + tcs.TrySetResult(cached); + return; + } + await FillGate.WaitAsync(); + acquired = true; + var started = Environment.TickCount64; + var result = await Client.Call(policy.Api, policy.Method, @params); var bytes = Encoding.UTF8.GetBytes(result is null ? "null" : JsJson.Stringify(result)); counter.RecordUpstream(Environment.TickCount64 - started); Cache.Set(key, bytes, policy.TtlMs); @@ -222,6 +245,7 @@ private static async Task Fill(MethodPolicy policy, JsonNode? @params, string ke } finally { + if (acquired) FillGate.Release(); InFlight.TryRemove(new KeyValuePair>(key, tcs.Task)); } } @@ -239,13 +263,18 @@ public static async Task Rpc(HttpContext ctx) var body = await ctx.ReadBody(); var api = body.Str("api"); var method = body.Str("method"); - if (api is null || method is null || !Allowlist.TryGetValue($"{api}.{method}", out var policy)) + // params must be present and structured (condenser methods take an + // array, bridge methods an object); the key and the upstream call are + // derived from that same node, so nothing is invented for either. + var @params = body.Field("params"); + if (api is null || method is null || @params is not (JsonObject or JsonArray) + || !Allowlist.TryGetValue($"{api}.{method}", out var policy)) { await Routes.Fallback(ctx); return; } - var resolution = await Resolve(policy, body.Field("params")); + var resolution = await Resolve(policy, @params); ctx.Response.Headers["X-Ssr-Cache"] = resolution.Outcome.ToString().ToUpperInvariant(); switch (resolution.Outcome) { @@ -285,7 +314,7 @@ public static async Task Stats(HttpContext ctx) ["coalesced"] = Interlocked.Read(ref c.Coalesced), ["error"] = Interlocked.Read(ref c.Error), ["timeout"] = Interlocked.Read(ref c.Timeout), - ["upstream_ms"] = Math.Round(c.UpstreamMs, 1), + ["upstream_ms"] = Math.Round(c.ReadUpstreamMs(), 1), }; } ctx.Response.Headers.CacheControl = "no-store"; diff --git a/dotnet/EcencyApi/Infrastructure/BytesCache.cs b/dotnet/EcencyApi/Infrastructure/BytesCache.cs index 7931071f..c9d14e24 100644 --- a/dotnet/EcencyApi/Infrastructure/BytesCache.cs +++ b/dotnet/EcencyApi/Infrastructure/BytesCache.cs @@ -27,6 +27,12 @@ private sealed class Entry private readonly LinkedList _lru = new(); private readonly object _lock = new(); private long _bytes; + private long _lastSweepMs; + + // Under pressure, expired entries anywhere in the list are dropped before a + // live one is evicted for space. A full pass is O(n), so it runs at most + // this often; lazy expiry on read covers the rest. + private const long SweepIntervalMs = 30_000; public BytesCache(long budgetBytes) { @@ -78,6 +84,10 @@ public void Set(string key, byte[] bytes, int ttlMs) var node = _lru.AddLast(key); _map[key] = new Entry { Bytes = bytes, ExpiresAtMs = NowMs + ttlMs, Node = node }; _bytes += bytes.Length; + if (_bytes > Budget) + { + SweepExpiredLocked(); + } while (_bytes > Budget && _lru.First is { } oldest && oldest != node) { RemoveLocked(oldest.Value, _map[oldest.Value]); @@ -85,6 +95,17 @@ public void Set(string key, byte[] bytes, int ttlMs) } } + private void SweepExpiredLocked() + { + var now = NowMs; + if (now - _lastSweepMs < SweepIntervalMs) return; + _lastSweepMs = now; + foreach (var (key, entry) in _map.Where(kv => kv.Value.ExpiresAtMs <= now).ToArray()) + { + RemoveLocked(key, entry); + } + } + private void RemoveLocked(string key, Entry entry) { _map.Remove(key); diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index 3b5cb015..ca68c629 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs @@ -335,7 +335,7 @@ public static class HiveClients // portfolio engine/chain layers came back empty for everyone. GetAccounts // also routes around such a node at runtime, but keeping them out of the pool // means correctness here does not depend on that fallback firing. - public static readonly string[] DefaultNodes = + public static readonly IReadOnlyList DefaultNodes = new[] { "https://api.hive.blog", "https://api.deathwing.me", @@ -347,5 +347,5 @@ public static class HiveClients "https://api.c0ff33a.uk", }; - public static readonly HiveRpcClient Default = new(DefaultNodes); + public static readonly HiveRpcClient Default = new(DefaultNodes.ToArray()); } From 132d5d2fe9823df005fc02f07dafa0f6e12a9559 Mon Sep 17 00:00:00 2001 From: feruzm Date: Fri, 21 Aug 2026 08:17:58 +0000 Subject: [PATCH 3/9] review: bound queued fills, hash the secret compare, partial classes, clone params Fills waiting for the gate are bounded too (SSR_RPC_MAX_QUEUED_FILLS, default 256; only fills actually waiting count), and a fill that waited past the budget is dropped instead of calling upstream for readers that have all given up. The secret comparison runs over SHA-256 digests so its timing is independent of the presented length. SsrRpc and Routes are static partial classes like the other handler hosts. The authorized handler path now has tests of its own (the digest is a replaceable seam), and they found a real defect: the request body's params node was placed into the RPC envelope directly and could not be re-parented, so every handler call answered 502. It travels as a clone. A null result is served as JSON null with a JSON content type; the route is an internal JSON contract, not a pipe of an upstream body, so the Express res.send quirks do not apply, and the handler says so. --- README.md | 1 + dotnet/EcencyApi.Tests/SsrRpcTests.cs | 102 +++++++++++++++++++++++++- dotnet/EcencyApi/Config.cs | 4 + dotnet/EcencyApi/Handlers/Routes.cs | 2 +- dotnet/EcencyApi/Handlers/SsrRpc.cs | 68 +++++++++++++++-- dotnet/docker-compose.yml | 2 + 6 files changed, 168 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 892cebdd..2598026a 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,7 @@ docker run -it --rm -p 4000:4000 \ | `SSR_RPC_BUDGET_MS` | wall-clock budget for one SSR RPC lookup before it answers 504 while the fill completes (default `1500`) | | `SSR_RPC_NODE_TIMEOUT_MS` | per-node timeout of the SSR RPC cache's own client, one attempt per node (default `1200`) | | `SSR_RPC_NODES` | comma-separated node pool for that client (default: the shared pool) | +| `SSR_RPC_MAX_FILLS` / `SSR_RPC_MAX_QUEUED_FILLS` | bound on upstream fills in progress (default `64`) and on fills waiting for that bound (default `256`); beyond the latter a miss fails fast | ## Swarm diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index f1d073b6..9f0bb4f7 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -26,6 +26,7 @@ private sealed class RpcStub : IAsyncDisposable public int Hits; public int DelayMs; public bool RpcError; + public bool NullResult; public RpcStub() { @@ -56,7 +57,9 @@ private async Task Loop() if (DelayMs > 0) await Task.Delay(DelayMs); var body = RpcError ? "{\"jsonrpc\":\"2.0\",\"id\":1,\"error\":{\"message\":\"stub error\"}}" - : "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"method\":\"" + method + "\",\"n\":" + n + ",\"text\":\"caf\\u00e9 \\ud83d\"}}"; + : NullResult + ? "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":null}" + : "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"method\":\"" + method + "\",\"n\":" + n + ",\"text\":\"caf\\u00e9 \\ud83d\"}}"; var bytes = Encoding.UTF8.GetBytes(body); ctx.Response.StatusCode = 200; ctx.Response.ContentType = "application/json"; @@ -76,12 +79,14 @@ public async ValueTask DisposeAsync() private static readonly SsrRpc.MethodPolicy Post = SsrRpc.Allowlist["bridge.get_post"]; private static readonly SsrRpc.MethodPolicy Props = SsrRpc.Allowlist["condenser_api.get_dynamic_global_properties"]; - private static void Use(RpcStub stub, long cacheBytes = 1 << 20, int budgetMs = 1500, int maxFills = 64) + private static void Use(RpcStub stub, long cacheBytes = 1 << 20, int budgetMs = 1500, int maxFills = 64, int maxQueued = 256) { SsrRpc.Client = new HiveRpcClient(new[] { stub.Url }, timeoutMs: 1000, failoverThreshold: 1); SsrRpc.Cache = new BytesCache(cacheBytes); SsrRpc.BudgetMs = budgetMs; SsrRpc.FillGate = new SemaphoreSlim(maxFills, maxFills); + SsrRpc.MaxQueuedFills = maxQueued; + SsrRpc.SecretDigest = null; SsrRpc.ResetForTests(); } @@ -234,6 +239,98 @@ public async Task Rpc_requires_structured_params_and_keys_the_call_on_them() Assert.Equal(2, stub.Hits); } + [Fact] + public async Task Queued_fills_are_bounded_and_a_fill_that_outlived_the_budget_in_the_queue_is_dropped() + { + await using var stub = new RpcStub { DelayMs = 400 }; + Use(stub, budgetMs: 100, maxFills: 1, maxQueued: 1); + // First fill holds the gate; second queues; third is over the queue bound. + var a = SsrRpc.Resolve(Post, P("q", "1")); + await Task.Delay(50); + var b = SsrRpc.Resolve(Post, P("q", "2")); + await Task.Delay(20); + var c = SsrRpc.Resolve(Post, P("q", "3")); + var ra = await a; var rb = await b; var rc = await c; + Assert.True(ra.Outcome == SsrRpc.Outcome.Timeout, $"a: {ra.Outcome} {ra.Error}"); + Assert.Equal(SsrRpc.Outcome.Timeout, rb.Outcome); + Assert.Equal(SsrRpc.Outcome.Unavailable, rc.Outcome); + Assert.Equal("fill queue full", rc.Error); + // The queued second fill waited past the budget for the gate, so it never + // calls upstream: one upstream hit in total, and its key is not cached. + await Task.Delay(900); + Assert.Equal(1, stub.Hits); + SsrRpc.BudgetMs = 1500; + Assert.Equal(SsrRpc.Outcome.Hit, (await SsrRpc.Resolve(Post, P("q", "1"))).Outcome); + Assert.Equal(SsrRpc.Outcome.Miss, (await SsrRpc.Resolve(Post, P("q", "2"))).Outcome); + } + + [Fact] + public async Task With_the_secret_configured_both_routes_serve_the_matching_header_and_nothing_else() + { + await using var stub = new RpcStub(); + Use(stub); + SsrRpc.SecretDigest = SsrRpc.Digest("right-secret"); + try + { + var ok = Request("POST", "/private-api/ssr/rpc", "right-secret", + "{\"api\":\"bridge\",\"method\":\"get_post\",\"params\":{\"author\":\"a\",\"permlink\":\"b\"}}"); + await SsrRpc.Rpc(ok); + Assert.True(ok.Response.StatusCode == 200, $"status {ok.Response.StatusCode}: {ResponseText(ok)}"); + Assert.Equal("MISS", ok.Response.Headers["X-Ssr-Cache"].ToString()); + Assert.StartsWith("{\"method\":\"get_post\"", ResponseText(ok)); + + var again = Request("POST", "/private-api/ssr/rpc", "right-secret", + "{\"api\":\"bridge\",\"method\":\"get_post\",\"params\":{\"permlink\":\"b\",\"author\":\"a\"}}"); + await SsrRpc.Rpc(again); + Assert.Equal("HIT", again.Response.Headers["X-Ssr-Cache"].ToString()); + Assert.Equal(1, stub.Hits); + + var stats = Request("GET", "/private-api/ssr/stats", "right-secret"); + await SsrRpc.Stats(stats); + Assert.Equal(200, stats.Response.StatusCode); + Assert.Contains("\"bridge.get_post\"", ResponseText(stats)); + Assert.Contains("\"hit\":1", ResponseText(stats)); + + foreach (var header in new[] { "wrong-secret", "right-secret-but-longer", "", null }) + { + var denied = Request("POST", "/private-api/ssr/rpc", header, + "{\"api\":\"bridge\",\"method\":\"get_post\",\"params\":{}}"); + await SsrRpc.Rpc(denied); + Assert.Equal(404, denied.Response.StatusCode); + var deniedStats = Request("GET", "/private-api/ssr/stats", header); + await SsrRpc.Stats(deniedStats); + Assert.Equal(200, deniedStats.Response.StatusCode); + Assert.DoesNotContain("methods", ResponseText(deniedStats)); + } + Assert.Equal(1, stub.Hits); + } + finally + { + SsrRpc.SecretDigest = null; + } + } + + [Fact] + public async Task A_null_result_is_served_as_json_null_with_a_json_content_type() + { + await using var stub = new RpcStub { NullResult = true }; + Use(stub); + SsrRpc.SecretDigest = SsrRpc.Digest("s"); + try + { + var ctx = Request("POST", "/private-api/ssr/rpc", "s", + "{\"api\":\"bridge\",\"method\":\"get_post\",\"params\":{\"author\":\"none\",\"permlink\":\"none\"}}"); + await SsrRpc.Rpc(ctx); + Assert.True(ctx.Response.StatusCode == 200, $"status {ctx.Response.StatusCode}: {ResponseText(ctx)}"); + Assert.StartsWith("application/json", ctx.Response.ContentType); + Assert.Equal("null", ResponseText(ctx)); + } + finally + { + SsrRpc.SecretDigest = null; + } + } + [Fact] public void Byte_budget_evicts_least_recently_used_and_refuses_oversize() { @@ -317,6 +414,7 @@ public async Task Without_the_secret_configured_both_routes_answer_like_unknown_ public void Authorized_requires_the_configured_secret_and_a_matching_header() { // With no secret configured nothing authorizes, header or not. + SsrRpc.SecretDigest = null; Assert.False(SsrRpc.Authorized(Request("POST", "/x", null))); Assert.False(SsrRpc.Authorized(Request("POST", "/x", ""))); Assert.False(SsrRpc.Authorized(Request("POST", "/x", "guess"))); diff --git a/dotnet/EcencyApi/Config.cs b/dotnet/EcencyApi/Config.cs index b0eccafa..46f04a43 100644 --- a/dotnet/EcencyApi/Config.cs +++ b/dotnet/EcencyApi/Config.cs @@ -61,6 +61,10 @@ public static class Config public static int SsrMaxConcurrentFills { get; } = int.TryParse(Env("SSR_RPC_MAX_FILLS"), out var f) && f > 0 ? f : 64; + // Upper bound on fills waiting for that gate; beyond it a miss fails fast. + public static int SsrMaxQueuedFills { get; } = + int.TryParse(Env("SSR_RPC_MAX_QUEUED_FILLS"), out var q) && q > 0 ? q : 256; + private static string? NonEmpty(string? value) => string.IsNullOrWhiteSpace(value) ? null : value.Trim(); private static string? Env(string name) => Environment.GetEnvironmentVariable(name); diff --git a/dotnet/EcencyApi/Handlers/Routes.cs b/dotnet/EcencyApi/Handlers/Routes.cs index 7d8068a0..bcff2f2c 100644 --- a/dotnet/EcencyApi/Handlers/Routes.cs +++ b/dotnet/EcencyApi/Handlers/Routes.cs @@ -16,7 +16,7 @@ namespace EcencyApi.Handlers; /// - unmatched GET/HEAD -> 200 + the template page; /// - unmatched other methods -> 404 + Express finalhandler "Cannot METHOD /path". /// -public static class Routes +public static partial class Routes { public static void Map(WebApplication app) { diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index e4498be1..7c3dfa2d 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -30,7 +30,7 @@ namespace EcencyApi.Handlers; /// answers 502. Either way the consumer falls back to its own node pool. /// - No per-request logging (invariant 6); counters on /stats instead. /// -public static class SsrRpc +public static partial class SsrRpc { internal sealed record MethodPolicy(string Api, string Method, int TtlMs) { @@ -58,6 +58,15 @@ internal sealed record MethodPolicy(string Api, string Method, int TtlMs) internal const string HeaderName = "X-Ecency-Internal"; + // The configured secret, held as a SHA-256 digest so the comparison below is + // over two equal-length values (FixedTimeEquals returns early on a length + // mismatch, which would otherwise leak the secret's length). Replaceable for + // tests, which cannot set process environment before Config initializes. + internal static byte[]? SecretDigest = Digest(Config.SsrInternalSecret); + + internal static byte[]? Digest(string? secret) => + secret is null ? null : SHA256.HashData(Encoding.UTF8.GetBytes(secret)); + // Replaceable for tests (loopback stub nodes, a small cache budget). internal static HiveRpcClient Client = new( Config.SsrRpcNodes ?? HiveClients.DefaultNodes.ToArray(), @@ -72,6 +81,11 @@ internal sealed record MethodPolicy(string Api, string Method, int TtlMs) // a slow pool plus many distinct keys must not pile up unbounded calls. internal static SemaphoreSlim FillGate = new(Config.SsrMaxConcurrentFills, Config.SsrMaxConcurrentFills); + // Bounds the fills WAITING for the gate as well: beyond this many, a new + // miss fails fast instead of queueing work no reader will wait for. + internal static int MaxQueuedFills = Config.SsrMaxQueuedFills; + private static int _queuedFills; + private static readonly ConcurrentDictionary> InFlight = new(); internal sealed class Counter @@ -106,13 +120,12 @@ internal static void ResetForTests() internal static bool Authorized(HttpContext ctx) { - var secret = Config.SsrInternalSecret; - if (secret is null) return false; + var expected = SecretDigest; + if (expected is null) return false; if (!ctx.Request.Headers.TryGetValue(HeaderName, out var values)) return false; var presented = values.ToString(); if (presented.Length == 0) return false; - return CryptographicOperations.FixedTimeEquals( - Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(secret)); + return CryptographicOperations.FixedTimeEquals(Digest(presented), expected); } // ---- key ----------------------------------------------------------------- @@ -215,10 +228,16 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa // The one upstream call behind a key. Runs to completion even when every // waiter has given up, so the cache still gets filled for the next reader. + internal sealed class FillRejectedException : Exception + { + public FillRejectedException(string message) : base(message) { } + } + private static async Task Fill(MethodPolicy policy, JsonNode @params, string key, Counter counter, TaskCompletionSource tcs) { var acquired = false; + var queued = false; try { // A reader can miss the cache, lose the race to another fill that @@ -229,10 +248,35 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key tcs.TrySetResult(cached); return; } - await FillGate.WaitAsync(); - acquired = true; + // A free slot is taken at once; otherwise the fill joins a bounded + // queue. Only fills actually waiting count against the bound. + if (FillGate.Wait(0)) + { + acquired = true; + } + else + { + if (Interlocked.Increment(ref _queuedFills) > MaxQueuedFills) + { + Interlocked.Decrement(ref _queuedFills); + throw new FillRejectedException("fill queue full"); + } + queued = true; + var enqueued = Environment.TickCount64; + await FillGate.WaitAsync(); + acquired = true; + // Every reader gave up at the budget while this sat in the queue; + // calling upstream now would only be stale traffic for nobody. + if (Environment.TickCount64 - enqueued > BudgetMs) + { + throw new FillRejectedException("fill expired in queue"); + } + } var started = Environment.TickCount64; - var result = await Client.Call(policy.Api, policy.Method, @params); + // Call() places params inside its own request envelope; a node that + // already hangs off the request body cannot be re-parented, so it + // travels as a clone. + var result = await Client.Call(policy.Api, policy.Method, @params.DeepClone()); var bytes = Encoding.UTF8.GetBytes(result is null ? "null" : JsJson.Stringify(result)); counter.RecordUpstream(Environment.TickCount64 - started); Cache.Set(key, bytes, policy.TtlMs); @@ -245,12 +289,20 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key } finally { + if (queued) Interlocked.Decrement(ref _queuedFills); if (acquired) FillGate.Release(); InFlight.TryRemove(new KeyValuePair>(key, tcs.Task)); } } // ---- handlers ------------------------------------------------------------ + // + // The body is the JSON serialization of the upstream `result`, always with + // an application/json content type: an object, an array, a string, a number + // or `null` exactly as JSON. This is a new internal contract read with + // res.json() by one consumer, not a pipe of an upstream HTTP body, so the + // Express res.send quirks the proxied routes preserve (null as an empty + // body, strings as text/html, numbers as text) do not apply here. public static async Task Rpc(HttpContext ctx) { diff --git a/dotnet/docker-compose.yml b/dotnet/docker-compose.yml index f2e61e02..829bd56c 100644 --- a/dotnet/docker-compose.yml +++ b/dotnet/docker-compose.yml @@ -35,6 +35,8 @@ services: - SSR_RPC_BUDGET_MS - SSR_RPC_NODE_TIMEOUT_MS - SSR_RPC_NODES + - SSR_RPC_MAX_FILLS + - SSR_RPC_MAX_QUEUED_FILLS ports: - "4000:4000" # Bound container logs so they can never grow unchecked on the host From 8215d381ceebdc387bf4c424a446843f310e1d34 Mon Sep 17 00:00:00 2001 From: feruzm Date: Fri, 21 Aug 2026 08:24:01 +0000 Subject: [PATCH 4/9] review: judge a queued fill by its last reader, not its enqueue time A queued fill is dropped only when no reader has attached within the budget. A fresh same-key reader that coalesces while the fill waits keeps it alive, so that reader is not refused for a timeout it never had. Test drives exactly that interleaving. --- dotnet/EcencyApi.Tests/SsrRpcTests.cs | 20 +++++++++++ dotnet/EcencyApi/Handlers/SsrRpc.cs | 48 ++++++++++++++++++--------- 2 files changed, 53 insertions(+), 15 deletions(-) diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 9f0bb4f7..4e23eaf1 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -264,6 +264,26 @@ public async Task Queued_fills_are_bounded_and_a_fill_that_outlived_the_budget_i Assert.Equal(SsrRpc.Outcome.Miss, (await SsrRpc.Resolve(Post, P("q", "2"))).Outcome); } + [Fact] + public async Task A_fresh_reader_that_coalesces_onto_a_queued_fill_keeps_it_alive() + { + await using var stub = new RpcStub { DelayMs = 200 }; + Use(stub, budgetMs: 150, maxFills: 1); + var a = SsrRpc.Resolve(Post, P("k", "1")); // holds the gate until ~200ms + await Task.Delay(10); + var b = SsrRpc.Resolve(Post, P("k", "2")); // queued at ~10ms, its reader gives up at ~160ms + await Task.Delay(180); + var c = SsrRpc.Resolve(Post, P("k", "2")); // fresh reader at ~190ms, coalesces + await Task.WhenAll(a, b, c); + Assert.Equal(SsrRpc.Outcome.Timeout, a.Result.Outcome); + Assert.Equal(SsrRpc.Outcome.Timeout, b.Result.Outcome); + // Judged by the fresh attach, not by the original enqueue: the fill ran. + await Task.Delay(500); + Assert.Equal(2, stub.Hits); + SsrRpc.BudgetMs = 1500; + Assert.Equal(SsrRpc.Outcome.Hit, (await SsrRpc.Resolve(Post, P("k", "2"))).Outcome); + } + [Fact] public async Task With_the_secret_configured_both_routes_serve_the_matching_header_and_nothing_else() { diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index 7c3dfa2d..07e704f2 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -86,7 +86,16 @@ internal sealed record MethodPolicy(string Api, string Method, int TtlMs) internal static int MaxQueuedFills = Config.SsrMaxQueuedFills; private static int _queuedFills; - private static readonly ConcurrentDictionary> InFlight = new(); + private sealed class Pending + { + public required Task Task; + // Last time a reader attached to this fill (created it or coalesced onto + // it). A queued fill is dropped only when nobody has attached within the + // budget, since every earlier reader has given up by then. + public long LastAttachMs = Environment.TickCount64; + } + + private static readonly ConcurrentDictionary InFlight = new(); internal sealed class Counter { @@ -183,27 +192,35 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa } var coalesced = true; - if (!InFlight.TryGetValue(key, out var pending)) + Pending pending; + if (InFlight.TryGetValue(key, out var existing)) + { + pending = existing; + Volatile.Write(ref pending.LastAttachMs, Environment.TickCount64); + } + else { var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - var winner = InFlight.GetOrAdd(key, tcs.Task); - if (ReferenceEquals(winner, tcs.Task)) + var created = new Pending { Task = tcs.Task }; + var winner = InFlight.GetOrAdd(key, created); + if (ReferenceEquals(winner, created)) { coalesced = false; - pending = tcs.Task; - _ = Fill(policy, @params, key, counter, tcs); + pending = created; + _ = Fill(policy, @params, key, counter, tcs, created); } else { pending = winner; + Volatile.Write(ref pending.LastAttachMs, Environment.TickCount64); } } if (coalesced) Interlocked.Increment(ref counter.Coalesced); else Interlocked.Increment(ref counter.Miss); - var finished = await Task.WhenAny(pending, Task.Delay(BudgetMs)); - if (!ReferenceEquals(finished, pending)) + var finished = await Task.WhenAny(pending.Task, Task.Delay(BudgetMs)); + if (!ReferenceEquals(finished, pending.Task)) { Interlocked.Increment(ref counter.Timeout); return new Resolution(Outcome.Timeout, Array.Empty(), "budget exceeded"); @@ -211,7 +228,7 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa try { - var bytes = await pending; + var bytes = await pending.Task; return new Resolution(coalesced ? Outcome.Coalesced : Outcome.Miss, bytes, null); } catch (HiveRpcClient.RpcException e) @@ -234,7 +251,7 @@ public FillRejectedException(string message) : base(message) { } } private static async Task Fill(MethodPolicy policy, JsonNode @params, string key, Counter counter, - TaskCompletionSource tcs) + TaskCompletionSource tcs, Pending pending) { var acquired = false; var queued = false; @@ -262,12 +279,13 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key throw new FillRejectedException("fill queue full"); } queued = true; - var enqueued = Environment.TickCount64; await FillGate.WaitAsync(); acquired = true; - // Every reader gave up at the budget while this sat in the queue; - // calling upstream now would only be stale traffic for nobody. - if (Environment.TickCount64 - enqueued > BudgetMs) + // Nobody has attached within the budget: every reader gave up + // while this sat in the queue, and calling upstream now would + // only be stale traffic for nobody. A fresh reader that coalesced + // in the meantime keeps the fill alive. + if (Environment.TickCount64 - Volatile.Read(ref pending.LastAttachMs) > BudgetMs) { throw new FillRejectedException("fill expired in queue"); } @@ -291,7 +309,7 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key { if (queued) Interlocked.Decrement(ref _queuedFills); if (acquired) FillGate.Release(); - InFlight.TryRemove(new KeyValuePair>(key, tcs.Task)); + InFlight.TryRemove(new KeyValuePair(key, pending)); } } From b2d5abdb318b6a8c14dd7131d7a63ee8f45078e0 Mon Sep 17 00:00:00 2001 From: feruzm Date: Fri, 21 Aug 2026 08:29:02 +0000 Subject: [PATCH 5/9] review: attach and expiry decide under one lock A reader attaching to a queued fill and the fill deciding it has expired were two unsynchronized steps, so a fresh reader could attach just after the stale timestamp was read and be refused within its own budget. Both now happen under the pending entry's lock: an expiring fill marks itself, and a reader that finds the mark clears the entry and starts a fresh fill. --- dotnet/EcencyApi/Handlers/SsrRpc.cs | 52 +++++++++++++++++++++++------ 1 file changed, 42 insertions(+), 10 deletions(-) diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index 07e704f2..4bdceb9b 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -91,8 +91,32 @@ private sealed class Pending public required Task Task; // Last time a reader attached to this fill (created it or coalesced onto // it). A queued fill is dropped only when nobody has attached within the - // budget, since every earlier reader has given up by then. + // budget, since every earlier reader has given up by then. Attaching and + // the expiry decision both happen under `lock (this)`, so a reader can + // never attach to a fill that has just decided to expire: it finds + // Expired set and starts a fresh fill instead. public long LastAttachMs = Environment.TickCount64; + public bool Expired; + + public bool TryAttach() + { + lock (this) + { + if (Expired) return false; + LastAttachMs = Environment.TickCount64; + return true; + } + } + + public bool TryExpire(int budgetMs) + { + lock (this) + { + if (Environment.TickCount64 - LastAttachMs <= budgetMs) return false; + Expired = true; + return true; + } + } } private static readonly ConcurrentDictionary InFlight = new(); @@ -193,13 +217,19 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa var coalesced = true; Pending pending; - if (InFlight.TryGetValue(key, out var existing)) - { - pending = existing; - Volatile.Write(ref pending.LastAttachMs, Environment.TickCount64); - } - else + while (true) { + if (InFlight.TryGetValue(key, out var existing)) + { + if (existing.TryAttach()) + { + pending = existing; + break; + } + // Expired while queued: clear it and start a fresh fill. + InFlight.TryRemove(new KeyValuePair(key, existing)); + continue; + } var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var created = new Pending { Task = tcs.Task }; var winner = InFlight.GetOrAdd(key, created); @@ -208,12 +238,14 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa coalesced = false; pending = created; _ = Fill(policy, @params, key, counter, tcs, created); + break; } - else + if (winner.TryAttach()) { pending = winner; - Volatile.Write(ref pending.LastAttachMs, Environment.TickCount64); + break; } + InFlight.TryRemove(new KeyValuePair(key, winner)); } if (coalesced) Interlocked.Increment(ref counter.Coalesced); @@ -285,7 +317,7 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key // while this sat in the queue, and calling upstream now would // only be stale traffic for nobody. A fresh reader that coalesced // in the meantime keeps the fill alive. - if (Environment.TickCount64 - Volatile.Read(ref pending.LastAttachMs) > BudgetMs) + if (pending.TryExpire(BudgetMs)) { throw new FillRejectedException("fill expired in queue"); } From fae645a5ddaf3f010eece14b5c0fc6e6a6c54781 Mon Sep 17 00:00:00 2001 From: feruzm Date: Fri, 21 Aug 2026 08:37:18 +0000 Subject: [PATCH 6/9] review: a reader whose fill expired before its wait began starts over once Attaching and starting the wait are two steps; a reader descheduled for longer than the budget between them can find the fill it attached to already judged expired. Its own budget has been spent on nothing yet, so it retries once from the cache instead of surfacing the rejection. --- dotnet/EcencyApi/Handlers/SsrRpc.cs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index 4bdceb9b..c5b5ed97 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -215,6 +215,8 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa return new Resolution(Outcome.Hit, cached, null); } + var retried = false; + again: var coalesced = true; Pending pending; while (true) @@ -263,6 +265,15 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa var bytes = await pending.Task; return new Resolution(coalesced ? Outcome.Coalesced : Outcome.Miss, bytes, null); } + catch (FillRejectedException) when (coalesced && !retried) + { + // The fill this reader attached to was judged expired (or refused) before + // the reader's own wait began, which can only happen if the reader was + // descheduled for longer than the budget between attaching and waiting. + // Its budget has not been spent on anything yet, so start over once. + retried = true; + goto again; + } catch (HiveRpcClient.RpcException e) { Interlocked.Increment(ref counter.Error); From ce751b42363db408651e2f4f8ba123a1456c77f9 Mon Sep 17 00:00:00 2001 From: feruzm Date: Fri, 21 Aug 2026 08:45:13 +0000 Subject: [PATCH 7/9] review: one wall-clock budget per lookup, retry included The retry after an expired fill waits only for what is left of the lookup's original budget, so the route never stays active past it. --- dotnet/EcencyApi/Handlers/SsrRpc.cs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index c5b5ed97..25711300 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -215,6 +215,8 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa return new Resolution(Outcome.Hit, cached, null); } + // One wall-clock budget for the whole lookup, retry included. + var deadline = Environment.TickCount64 + BudgetMs; var retried = false; again: var coalesced = true; @@ -253,7 +255,8 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa if (coalesced) Interlocked.Increment(ref counter.Coalesced); else Interlocked.Increment(ref counter.Miss); - var finished = await Task.WhenAny(pending.Task, Task.Delay(BudgetMs)); + var remaining = (int)Math.Max(0, deadline - Environment.TickCount64); + var finished = await Task.WhenAny(pending.Task, Task.Delay(remaining)); if (!ReferenceEquals(finished, pending.Task)) { Interlocked.Increment(ref counter.Timeout); From dde07c783dffc88f18c07381ec408969fa10bec3 Mon Sep 17 00:00:00 2001 From: feruzm Date: Fri, 21 Aug 2026 09:09:22 +0000 Subject: [PATCH 8/9] review: purge expired entries on every over-budget set The throttled full pass let an entry that expired inside the 30s window outlive a live one under pressure. The cache keeps an expiry-ordered index alongside the LRU, so every over-budget Set drops every expired entry first at O(log n) apiece, and only then evicts from the LRU head. Test covers two expiries in quick succession under pressure. --- dotnet/EcencyApi.Tests/SsrRpcTests.cs | 22 +++++++++++++++ dotnet/EcencyApi/Infrastructure/BytesCache.cs | 27 ++++++++++--------- 2 files changed, 36 insertions(+), 13 deletions(-) diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 4e23eaf1..7baed6ce 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -221,6 +221,28 @@ public async Task Under_pressure_expired_entries_go_before_live_ones() Assert.True(cache.TryGet("new", out _)); } + [Fact] + public async Task Every_over_budget_set_purges_expired_entries_first_not_only_once_in_a_while() + { + var cache = new BytesCache(120); + cache.Set("live-a", new byte[40], 60_000); + cache.Set("short-1", new byte[40], 50); + await Task.Delay(120); + cache.Set("live-b", new byte[40], 60_000); // over budget: short-1 (expired) goes + Assert.False(cache.TryGet("short-1", out _)); + Assert.True(cache.TryGet("live-a", out _)); + // A second expiry moments later, then more pressure: it must go before + // the live entry at the LRU head, even though a purge just ran. + cache.Set("short-2", new byte[40], 50); + await Task.Delay(120); + cache.Set("live-c", new byte[40], 60_000); + Assert.False(cache.TryGet("short-2", out _)); + Assert.True(cache.TryGet("live-a", out _)); + Assert.True(cache.TryGet("live-b", out _)); + Assert.True(cache.TryGet("live-c", out _)); + Assert.Equal(120, cache.Bytes); + } + [Fact] public async Task Rpc_requires_structured_params_and_keys_the_call_on_them() { diff --git a/dotnet/EcencyApi/Infrastructure/BytesCache.cs b/dotnet/EcencyApi/Infrastructure/BytesCache.cs index c9d14e24..6f77454f 100644 --- a/dotnet/EcencyApi/Infrastructure/BytesCache.cs +++ b/dotnet/EcencyApi/Infrastructure/BytesCache.cs @@ -25,14 +25,11 @@ private sealed class Entry private readonly Dictionary _map = new(); // Head = least recently used, tail = most recently used. private readonly LinkedList _lru = new(); + // Expiry order, so that under pressure every expired entry anywhere in + // the list is dropped before a live one is evicted, at O(log n) apiece. + private readonly SortedSet<(long ExpiresAtMs, string Key)> _byExpiry = new(); private readonly object _lock = new(); private long _bytes; - private long _lastSweepMs; - - // Under pressure, expired entries anywhere in the list are dropped before a - // live one is evicted for space. A full pass is O(n), so it runs at most - // this often; lazy expiry on read covers the rest. - private const long SweepIntervalMs = 30_000; public BytesCache(long budgetBytes) { @@ -82,11 +79,13 @@ public void Set(string key, byte[] bytes, int ttlMs) RemoveLocked(key, existing); } var node = _lru.AddLast(key); - _map[key] = new Entry { Bytes = bytes, ExpiresAtMs = NowMs + ttlMs, Node = node }; + var entry = new Entry { Bytes = bytes, ExpiresAtMs = NowMs + ttlMs, Node = node }; + _map[key] = entry; + _byExpiry.Add((entry.ExpiresAtMs, key)); _bytes += bytes.Length; if (_bytes > Budget) { - SweepExpiredLocked(); + PurgeExpiredLocked(); } while (_bytes > Budget && _lru.First is { } oldest && oldest != node) { @@ -95,14 +94,15 @@ public void Set(string key, byte[] bytes, int ttlMs) } } - private void SweepExpiredLocked() + // Every expired entry, wherever it sits in the LRU, goes before any live one. + private void PurgeExpiredLocked() { var now = NowMs; - if (now - _lastSweepMs < SweepIntervalMs) return; - _lastSweepMs = now; - foreach (var (key, entry) in _map.Where(kv => kv.Value.ExpiresAtMs <= now).ToArray()) + while (_byExpiry.Count > 0) { - RemoveLocked(key, entry); + var (expiresAt, key) = _byExpiry.Min; + if (expiresAt > now) break; + RemoveLocked(key, _map[key]); } } @@ -110,6 +110,7 @@ private void RemoveLocked(string key, Entry entry) { _map.Remove(key); _lru.Remove(entry.Node); + _byExpiry.Remove((entry.ExpiresAtMs, key)); _bytes -= entry.Bytes.Length; } } From 806b4c2be5fbb678a004ad05aff005319e0e8ea7 Mon Sep 17 00:00:00 2001 From: feruzm Date: Fri, 21 Aug 2026 09:14:53 +0000 Subject: [PATCH 9/9] review: ordinal key order in the expiry index The index compares keys ordinally, as the dictionary does, so two distinct keys that a culture-aware comparison could collate as equal cannot desynchronize the two structures; the purge also tolerates an index entry without a map entry rather than throwing under the lock. --- dotnet/EcencyApi.Tests/SsrRpcTests.cs | 17 +++++++++++++++ dotnet/EcencyApi/Infrastructure/BytesCache.cs | 21 +++++++++++++++++-- 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 7baed6ce..13a7afbe 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -243,6 +243,23 @@ public async Task Every_over_budget_set_purges_expired_entries_first_not_only_on Assert.Equal(120, cache.Bytes); } + [Fact] + public async Task Keys_that_share_an_expiry_millisecond_are_indexed_and_purged_independently() + { + var cache = new BytesCache(100); + // Same TTL set back to back, keys that differ only in characters a + // culture-aware comparison can treat as ignorable. + cache.Set("k-a", new byte[30], 50); + cache.Set("k-\u00ADa", new byte[30], 50); + cache.Set("k-A", new byte[30], 50); + Assert.Equal(3, cache.Count); + await Task.Delay(120); + cache.Set("live", new byte[40], 60_000); // over budget: all three expired must go + Assert.Equal(1, cache.Count); + Assert.Equal(40, cache.Bytes); + Assert.True(cache.TryGet("live", out _)); + } + [Fact] public async Task Rpc_requires_structured_params_and_keys_the_call_on_them() { diff --git a/dotnet/EcencyApi/Infrastructure/BytesCache.cs b/dotnet/EcencyApi/Infrastructure/BytesCache.cs index 6f77454f..26fce2df 100644 --- a/dotnet/EcencyApi/Infrastructure/BytesCache.cs +++ b/dotnet/EcencyApi/Infrastructure/BytesCache.cs @@ -27,7 +27,17 @@ private sealed class Entry private readonly LinkedList _lru = new(); // Expiry order, so that under pressure every expired entry anywhere in // the list is dropped before a live one is evicted, at O(log n) apiece. - private readonly SortedSet<(long ExpiresAtMs, string Key)> _byExpiry = new(); + // Keys are compared ordinally, the same way the dictionary compares them: + // they carry client-supplied JSON, and a culture-sensitive tie-break could + // collate two distinct keys as equal and desynchronize the two structures. + private static readonly IComparer<(long ExpiresAtMs, string Key)> ExpiryOrder = + Comparer<(long ExpiresAtMs, string Key)>.Create((a, b) => + { + var byTime = a.ExpiresAtMs.CompareTo(b.ExpiresAtMs); + return byTime != 0 ? byTime : string.CompareOrdinal(a.Key, b.Key); + }); + + private readonly SortedSet<(long ExpiresAtMs, string Key)> _byExpiry = new(ExpiryOrder); private readonly object _lock = new(); private long _bytes; @@ -102,7 +112,14 @@ private void PurgeExpiredLocked() { var (expiresAt, key) = _byExpiry.Min; if (expiresAt > now) break; - RemoveLocked(key, _map[key]); + if (_map.TryGetValue(key, out var entry)) + { + RemoveLocked(key, entry); + } + else + { + _byExpiry.Remove((expiresAt, key)); + } } }