From 3e42427711f7cac3aa8b65227c8c49b791f852c6 Mon Sep 17 00:00:00 2001 From: feruzm Date: Thu, 12 Nov 2026 06:10:46 +0000 Subject: [PATCH 1/5] SSR read cache: allow vote-free bridge v2 reads with per-node v1 fallback Allowlist bridge.get_post_v2, get_ranked_posts_v2, get_account_posts_v2 and get_discussion_v2 with their v1 TTL and class, plus bridge.list_votes (cheap, get_post TTL). A node that answers `unknown method` for a v2 read is asked the v1 twin at once, in the same attempt, with the v2-only top_votes param dropped (v1 rejects it). It is not failed over and not counted as a failure, and it is asked v1 directly for 30 minutes, after which v2 is tried again. The answer is cached under the requested key as returned; the consumer normalizes either shape. /stats lists each node's current marks. Closes #119 --- dotnet/EcencyApi.Tests/SsrRpcTests.cs | 197 +++++++++++++++++- dotnet/EcencyApi/Handlers/SsrRpc.cs | 25 ++- .../EcencyApi/Infrastructure/HiveRpcClient.cs | 85 +++++++- 3 files changed, 292 insertions(+), 15 deletions(-) diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 7741ca6e..ce9b4163 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -28,6 +28,11 @@ private sealed class RpcStub : IAsyncDisposable public bool RpcError; public bool NullResult; public volatile string LastMethod = ""; + // Methods this node answers `unknown method` for, the way a hivemind + // without the v2 reads does (HTTP 200, code -32602, assert message). + public HashSet Unknown = new(); + public readonly System.Collections.Concurrent.ConcurrentQueue Methods = new(); + public readonly System.Collections.Concurrent.ConcurrentQueue Params = new(); public RpcStub() { @@ -72,8 +77,19 @@ private async Task Loop() method = rawMethod.Contains('.') ? rawMethod[(rawMethod.IndexOf('.') + 1)..] : rawMethod; } LastMethod = rawMethod; + Methods.Enqueue(rawMethod); + Params.Enqueue(req?["params"]?.ToJsonString() ?? ""); if (DelayMs > 0) await Task.Delay(DelayMs); - var body = RpcError || legacyBridge + // Real nodes: the v1 bridge reads reject the v2-only `top_votes` + // param ("got an unexpected keyword argument"). + var unexpectedArg = rawMethod.StartsWith("bridge.") && !rawMethod.EndsWith("_v2") + && req?["params"] is JsonObject po && po.ContainsKey("top_votes"); + var assertion = unexpectedArg ? "got an unexpected keyword argument 'top_votes'" + : Unknown.Contains(rawMethod) ? "unknown method: bridge_api" + method : null; + var body = assertion is not null + ? "{\"jsonrpc\":\"2.0\",\"id\":1,\"error\":{\"code\":-32602,\"message\":\"Assert Exception:" + assertion + + "\",\"data\":{\"code\":10,\"extension\":{\"assertion_expression\":\"" + assertion + "\"}}}}" + : RpcError || legacyBridge ? "{\"jsonrpc\":\"2.0\",\"id\":1,\"error\":{\"message\":\"" + (legacyBridge ? "Assert Exception:api_itr != data._registered_apis.end(): Could not find API bridge" : "stub error") + "\"}}" : NullResult ? "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":null}" @@ -624,6 +640,8 @@ public void Allowlist_is_read_only_and_names_every_method_the_consumer_routes() "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", + "bridge.get_post_v2", "bridge.get_ranked_posts_v2", "bridge.get_account_posts_v2", + "bridge.get_discussion_v2", "bridge.list_votes", }) { Assert.True(SsrRpc.Allowlist.ContainsKey(key), key); @@ -646,7 +664,11 @@ public void Allowlist_classifies_the_feed_shaped_reads_as_heavy() .OrderBy(k => k, StringComparer.Ordinal) .ToArray(); Assert.Equal( - new[] { "bridge.get_account_posts", "bridge.get_discussion", "bridge.get_ranked_posts" }, + new[] + { + "bridge.get_account_posts", "bridge.get_account_posts_v2", "bridge.get_discussion", + "bridge.get_discussion_v2", "bridge.get_ranked_posts", "bridge.get_ranked_posts_v2", + }, heavy); } @@ -713,4 +735,175 @@ public async Task Call_classes_off_files_every_read_under_the_cheap_profile() Assert.Equal(1, view["samples"]!.GetValue()); Assert.Equal(0, view["heavy_samples"]!.GetValue()); } + + // ---- vote-free bridge v2 reads with a per-node v1 fallback (#119) -------- + + private static SsrRpc.MethodPolicy Policy(string key) + { + Assert.True(SsrRpc.Allowlist.TryGetValue(key, out var policy), $"{key} is not allowlisted"); + return policy!; + } + + private static JsonObject PV2(string author, string permlink) => + new() { ["author"] = author, ["permlink"] = permlink, ["top_votes"] = 0 }; + + [Fact] + public async Task A_node_that_serves_get_post_v2_answers_v2_and_the_result_is_cached() + { + await using var stub = new RpcStub(); + Use(stub); + var v2 = Policy("bridge.get_post_v2"); + + var first = await SsrRpc.Resolve(v2, PV2("a", "b")); + Assert.Equal(SsrRpc.Outcome.Miss, first.Outcome); + Assert.StartsWith("{\"method\":\"get_post_v2\"", Encoding.UTF8.GetString(first.Bytes)); + Assert.Equal(new[] { "bridge.get_post_v2" }, stub.Methods.ToArray()); + // The v2 params travel as given, top_votes included. + Assert.Contains("\"top_votes\":0", stub.Params.Single()); + + var second = await SsrRpc.Resolve(v2, PV2("a", "b")); + Assert.Equal(SsrRpc.Outcome.Hit, second.Outcome); + Assert.Equal(first.Bytes, second.Bytes); + Assert.Equal(1, stub.Hits); + } + + [Fact] + public async Task Unknown_method_retries_the_same_node_with_v1_without_failing_it_over_and_then_goes_straight_to_v1() + { + await using var lacking = new RpcStub { Unknown = { "bridge.get_post_v2" } }; + await using var other = new RpcStub(); + Use(lacking); + SsrRpc.Client = new HiveRpcClient(new[] { lacking.Url, other.Url }, timeoutMs: 1000, failoverThreshold: 1); + var v2 = Policy("bridge.get_post_v2"); + + var r = await SsrRpc.Resolve(v2, PV2("a", "b")); + Assert.True(r.Outcome == SsrRpc.Outcome.Miss, $"{r.Outcome}: {r.Error}"); + // The v1 answer, from the same node, with the v2-only param dropped (a + // v1 node rejects it as an unexpected keyword argument). + Assert.StartsWith("{\"method\":\"get_post\"", Encoding.UTF8.GetString(r.Bytes)); + Assert.Equal(new[] { "bridge.get_post_v2", "bridge.get_post" }, lacking.Methods.ToArray()); + Assert.DoesNotContain("top_votes", lacking.Params.Last()); + Assert.Equal(0, other.Hits); + var health = SsrRpc.Client.HealthSnapshot()[0]!; + Assert.Equal(0, health["failures"]!.GetValue()); + Assert.Equal(0, health["consecutive_failures"]!.GetValue()); + + // Cached under the requested (v2) key as returned. + var hit = await SsrRpc.Resolve(v2, PV2("a", "b")); + Assert.Equal(SsrRpc.Outcome.Hit, hit.Outcome); + Assert.Equal(r.Bytes, hit.Bytes); + + // The node is now known to lack v2: the next miss goes straight to v1. + var next = await SsrRpc.Resolve(v2, PV2("a", "c")); + Assert.Equal(SsrRpc.Outcome.Miss, next.Outcome); + Assert.Equal(new[] { "bridge.get_post_v2", "bridge.get_post", "bridge.get_post" }, lacking.Methods.ToArray()); + Assert.Equal(0, other.Hits); + } + + [Fact] + public async Task An_error_from_the_v1_retry_is_an_rpc_error_not_a_node_failure() + { + await using var stub = new RpcStub { Unknown = { "bridge.get_post_v2", "bridge.get_post" } }; + Use(stub); + var r = await SsrRpc.Resolve(Policy("bridge.get_post_v2"), PV2("a", "b")); + Assert.Equal(SsrRpc.Outcome.RpcError, r.Outcome); + Assert.Equal(new[] { "bridge.get_post_v2", "bridge.get_post" }, stub.Methods.ToArray()); + Assert.Equal(0, SsrRpc.Client.HealthSnapshot()[0]!["failures"]!.GetValue()); + } + + [Fact] + public async Task Other_rpc_errors_on_a_v2_read_do_not_mark_the_node_or_retry_v1() + { + await using var stub = new RpcStub { RpcError = true }; + Use(stub); + var v2 = Policy("bridge.get_post_v2"); + Assert.Equal(SsrRpc.Outcome.RpcError, (await SsrRpc.Resolve(v2, PV2("a", "b"))).Outcome); + Assert.Equal(new[] { "bridge.get_post_v2" }, stub.Methods.ToArray()); + stub.RpcError = false; + Assert.Equal(SsrRpc.Outcome.Miss, (await SsrRpc.Resolve(v2, PV2("a", "c"))).Outcome); + Assert.Equal(new[] { "bridge.get_post_v2", "bridge.get_post_v2" }, stub.Methods.ToArray()); + } + + [Fact] + public async Task Once_the_mark_has_lapsed_v2_is_tried_first_again() + { + await using var stub = new RpcStub { Unknown = { "bridge.get_post_v2" } }; + Use(stub); + long now = 1_000_000; + SsrRpc.Client = new HiveRpcClient(new[] { stub.Url }, 1000, 1, () => now); + var v2 = Policy("bridge.get_post_v2"); + + await SsrRpc.Resolve(v2, PV2("a", "1")); + Assert.Equal(new[] { "bridge.get_post_v2", "bridge.get_post" }, stub.Methods.ToArray()); + + // Still inside the mark: straight to v1. + now += HiveRpcClient.UnsupportedMethodTtlMs - 1; + await SsrRpc.Resolve(v2, PV2("a", "2")); + Assert.Equal("bridge.get_post", stub.Methods.Last()); + Assert.Equal(3, stub.Hits); + + // The node has been upgraded meanwhile; past the mark it is asked v2 again. + stub.Unknown.Clear(); + now += 2; + var r = await SsrRpc.Resolve(v2, PV2("a", "3")); + Assert.Equal(SsrRpc.Outcome.Miss, r.Outcome); + Assert.Equal("bridge.get_post_v2", stub.Methods.Last()); + Assert.Equal(4, stub.Hits); + Assert.StartsWith("{\"method\":\"get_post_v2\"", Encoding.UTF8.GetString(r.Bytes)); + } + + [Fact] + public async Task Stats_list_the_v2_methods_and_list_votes_with_their_class_and_the_node_marks() + { + await using var stub = new RpcStub { Unknown = { "bridge.get_ranked_posts_v2" } }; + Use(stub); + SsrRpc.SecretDigest = SsrRpc.Digest("right-secret"); + try + { + await SsrRpc.Resolve(Policy("bridge.get_post_v2"), PV2("a", "b")); + await SsrRpc.Resolve(Policy("bridge.get_ranked_posts_v2"), new JsonObject { ["sort"] = "trending", ["tag"] = "" }); + await SsrRpc.Resolve(Policy("bridge.get_account_posts_v2"), new JsonObject { ["sort"] = "posts", ["account"] = "a" }); + await SsrRpc.Resolve(Policy("bridge.get_discussion_v2"), P("a", "b")); + await SsrRpc.Resolve(Policy("bridge.list_votes"), new JsonObject { ["author"] = "a", ["permlink"] = "b", ["limit"] = 100 }); + + var stats = Request("GET", "/private-api/ssr/stats", "right-secret"); + await SsrRpc.Stats(stats); + var body = JsonNode.Parse(ResponseText(stats))!; + var methods = body["methods"]!; + Assert.Equal("cheap", methods["bridge.get_post_v2"]?["class"]?.GetValue()); + Assert.Equal("heavy", methods["bridge.get_ranked_posts_v2"]?["class"]?.GetValue()); + Assert.Equal("heavy", methods["bridge.get_account_posts_v2"]?["class"]?.GetValue()); + Assert.Equal("heavy", methods["bridge.get_discussion_v2"]?["class"]?.GetValue()); + Assert.Equal("cheap", methods["bridge.list_votes"]?["class"]?.GetValue()); + Assert.Equal(1, methods["bridge.get_ranked_posts_v2"]?["miss"]?.GetValue()); + + // Which node lacks which v2 read, so the fallback can be retired + // once the list stays empty. + var node = Assert.Single(body["nodes"]!.AsArray())!; + var unsupported = node["unsupported"]?.AsArray().Select(m => m!.GetValue()).ToArray(); + Assert.Equal(new[] { "bridge.get_ranked_posts_v2" }, unsupported); + } + finally + { + SsrRpc.SecretDigest = null; + } + } + + [Fact] + public void V2_twins_take_the_v1_ttl_and_class_and_list_votes_is_a_cheap_point_read() + { + foreach (var v1 in new[] { "bridge.get_post", "bridge.get_ranked_posts", "bridge.get_account_posts", "bridge.get_discussion" }) + { + var one = Policy(v1); + var two = Policy(v1 + "_v2"); + Assert.Equal(one.TtlMs, two.TtlMs); + Assert.Equal(one.Class, two.Class); + Assert.Equal(one.Method, two.Fallback); + Assert.Null(one.Fallback); + } + var votes = Policy("bridge.list_votes"); + Assert.Equal(CallClass.Cheap, votes.Class); + Assert.Equal(Policy("bridge.get_post").TtlMs, votes.TtlMs); + Assert.Null(votes.Fallback); + } } diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index 506d57d9..efed6dd3 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -38,7 +38,7 @@ public static partial class SsrRpc /// between nodes, so the pool is ordered for them from their own latency /// profile (see ). Required, not defaulted: a new /// method has to be classified, not silently filed as cheap. - internal sealed record MethodPolicy(string Api, string Method, int TtlMs, CallClass Class) + internal sealed record MethodPolicy(string Api, string Method, int TtlMs, CallClass Class, string? Fallback = null) { public string Key => $"{Api}.{Method}"; } @@ -57,6 +57,14 @@ internal sealed record MethodPolicy(string Api, string Method, int TtlMs, CallCl new MethodPolicy("bridge", "get_account_posts", 30_000, CallClass.Heavy), new MethodPolicy("bridge", "get_discussion", 30_000, CallClass.Heavy), new MethodPolicy("bridge", "get_post", 30_000, CallClass.Cheap), + // Vote-free v2 twins: the v1 TTL and class, plus the v1 method a node + // without them is asked instead (see HiveRpcClient.CallMethod). A post's + // votes come separately from list_votes, a point read on the get_post TTL. + new MethodPolicy("bridge", "get_ranked_posts_v2", 15_000, CallClass.Heavy, Fallback: "get_ranked_posts"), + new MethodPolicy("bridge", "get_account_posts_v2", 30_000, CallClass.Heavy, Fallback: "get_account_posts"), + new MethodPolicy("bridge", "get_discussion_v2", 30_000, CallClass.Heavy, Fallback: "get_discussion"), + new MethodPolicy("bridge", "get_post_v2", 30_000, CallClass.Cheap, Fallback: "get_post"), + new MethodPolicy("bridge", "list_votes", 30_000, CallClass.Cheap), new MethodPolicy("bridge", "get_profile", 60_000, CallClass.Cheap), new MethodPolicy("bridge", "get_profiles", 60_000, CallClass.Cheap), new MethodPolicy("bridge", "get_community", 300_000, CallClass.Cheap), @@ -320,6 +328,18 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa } } + // A v2 read answered by its v1 twin: the same params minus the v2-only + // `top_votes`, which a v1 node rejects as an unexpected keyword argument. + // The answer is cached under the requested (v2) key as returned, in either + // shape: the route stays a raw pass-through and the consumer's normalizer + // takes both, so one fill per key still serves every reader of that key. + internal static JsonNode V1Params(JsonNode @params) + { + var copy = @params.DeepClone(); + if (copy is JsonObject obj) obj.Remove("top_votes"); + return copy; + } + // 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 @@ -377,7 +397,8 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key // hivemind. The node already hangs off the request body and cannot be // re-parented into the envelope, so it travels as a clone. result = await Client.CallMethod($"{policy.Api}.{policy.Method}", @params.DeepClone(), - callClass: CallClasses ? policy.Class : CallClass.Cheap); + callClass: CallClasses ? policy.Class : CallClass.Cheap, + fallback: policy.Fallback is { } v1 ? ($"{policy.Api}.{v1}", V1Params(@params)) : null); } finally { diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index 40be5b1d..978ced74 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using System.Text; using System.Text.Json.Nodes; @@ -13,6 +14,8 @@ namespace EcencyApi.Infrastructure; /// (dhive semantics — an application error, not an unhealthy node). /// - Overload statuses (429/502/503/504) advance to the next node immediately /// instead of burning a same-node retry. +/// - An `unknown method` answer to a call that names a fallback method asks +/// the same node the fallback, and marks the node for a while (CallMethod). /// /// Latency is tracked per call class (see ) because the /// reads made through this client are bimodal in cost. Failure state is not: it @@ -27,8 +30,13 @@ public sealed class HiveRpcClient private readonly int _timeoutMs; private readonly int _failoverThreshold; private readonly NodeHealthTracker _health; + private readonly Func _clock; private int _seq; + // Per node, the methods it answered `unknown method` for and until when (on + // _clock). Bounded by nodes x fallback-capable methods; see CallMethod. + private readonly ConcurrentDictionary<(int Node, string Method), long> _unsupportedUntil = new(); + // timeoutMs 2000 / failoverThreshold 2 mirror the dhive Client options the // Node service constructed its clients with. public HiveRpcClient(string[] nodes, int timeoutMs = 2000, int failoverThreshold = 2) @@ -43,6 +51,7 @@ internal HiveRpcClient(string[] nodes, int timeoutMs, int failoverThreshold, Fun _timeoutMs = timeoutMs; _failoverThreshold = Math.Max(1, failoverThreshold); _health = new NodeHealthTracker(nodes.Length, clock); + _clock = clock ?? (() => Environment.TickCount64); } public IReadOnlyList Nodes => _nodes; @@ -82,6 +91,10 @@ public JsonArray HealthSnapshot() node["rate_limited_for_ms"] = v.RateLimitedForMs; node["parked_for_ms"] = v.FailureParkedForMs; node["failure_rate"] = Math.Round(v.FailureRate, 3); + // Methods this node is asked the fallback for until its mark lapses. + var unsupported = new JsonArray(); + foreach (var m in UnsupportedMethods(v.Index)) unsupported.Add(m); + node["unsupported"] = unsupported; arr.Add(node); } return arr; @@ -98,6 +111,13 @@ public RpcException(string message) : base(message) { } /// preference before the first well-formed answer is accepted as-is. private const int MaxPreferenceProbes = 2; + /// How long a node that answered `unknown method` is asked the + /// fallback instead. Long enough that a node lacking the method costs one + /// wasted round trip per method per half hour, short enough that a node + /// upgraded in place is used for the new method again within the half hour, + /// with no restart. + internal const long UnsupportedMethodTtlMs = 30 * 60 * 1000; + // ---- calls ------------------------------------------------------------- /// Optional shape check for the RPC result. A node @@ -145,29 +165,59 @@ public RpcException(string message) : base(message) { } /// as given, which jussi/HAF route to hived or hivemind by prefix. Needed /// for every hivemind (`bridge`) read; works for condenser_api too. /// + /// An older method serving the same read, for nodes + /// that do not have this one yet. A node that answers `unknown method` is not + /// failed over: it is asked the fallback at once, in the same attempt, and is + /// asked the fallback directly for . + /// Whatever that node answers is returned as is. public Task CallMethod(string qualifiedMethod, JsonNode @params, Func? validateResult = null, Func? preferResult = null, - CallClass callClass = CallClass.Cheap) + CallClass callClass = CallClass.Cheap, + (string Method, JsonNode Params)? fallback = null) { - var request = new JsonObject - { - ["id"] = Interlocked.Increment(ref _seq), - ["jsonrpc"] = "2.0", - ["method"] = qualifiedMethod, - ["params"] = @params, - }; - return Send(request, qualifiedMethod, validateResult, preferResult, callClass); + var request = Envelope(qualifiedMethod, @params); + var fallbackRequest = fallback is { } f ? Envelope(f.Method, f.Params) : null; + return Send(request, qualifiedMethod, validateResult, preferResult, callClass, fallbackRequest); + } + + private JsonObject Envelope(string qualifiedMethod, JsonNode @params) => new() + { + ["id"] = Interlocked.Increment(ref _seq), + ["jsonrpc"] = "2.0", + ["method"] = qualifiedMethod, + ["params"] = @params, + }; + + // hivemind without the method: HTTP 200, code -32602, "Assert + // Exception:unknown method: bridge_apiget_post_v2" (verified 2026-11-12). + private static bool IsUnknownMethod(RpcException e) => + e.Message.Contains("unknown method", StringComparison.OrdinalIgnoreCase); + + private bool Unsupported(int nodeIndex, string method) + { + if (!_unsupportedUntil.TryGetValue((nodeIndex, method), out var until)) return false; + if (_clock() < until) return true; + _unsupportedUntil.TryRemove(new KeyValuePair<(int, string), long>((nodeIndex, method), until)); + return false; } + private IEnumerable UnsupportedMethods(int nodeIndex) => + _unsupportedUntil.Keys.Where(k => k.Node == nodeIndex && Unsupported(nodeIndex, k.Method)) + .Select(k => k.Method) + .OrderBy(m => m, StringComparer.Ordinal) + .ToArray(); + private async Task Send(JsonObject request, string method, Func? validateResult, Func? preferResult, - CallClass callClass) + CallClass callClass, + JsonObject? fallbackRequest = null) { // JsJson: a lone-surrogate username from a client token must serialize // (JSON.stringify semantics) instead of throwing in the writer. var body = JsJson.Stringify(request); + var fallbackBody = fallbackRequest is null ? null : JsJson.Stringify(fallbackRequest); Exception? lastError = null; JsonNode? unpreferred = null; @@ -185,7 +235,20 @@ public RpcException(string message) : base(message) { } var started = NowMs; try { - var result = await CallNode(node, body); + var useFallback = fallbackBody != null && Unsupported(nodeIndex, method); + JsonNode? result; + try + { + result = await CallNode(node, useFallback ? fallbackBody! : body); + } + catch (RpcException e) when (fallbackBody != null && !useFallback && IsUnknownMethod(e)) + { + // The node is up, it just predates the method: same node, + // older method, and remember it. Not a failure, not a failover. + _unsupportedUntil[(nodeIndex, method)] = _clock() + UnsupportedMethodTtlMs; + started = NowMs; + result = await CallNode(node, fallbackBody); + } if (validateResult != null && !validateResult(result)) { // Don't burn a same-node retry: a node serving malformed From 402655cfcd202b469fcdc7046fd5f8351c5df000 Mon Sep 17 00:00:00 2001 From: feruzm Date: Thu, 12 Nov 2026 06:22:37 +0000 Subject: [PATCH 2/5] SSR read cache: harden the v2 unknown-method fallback - Mark a node only for an assert that starts with "Assert Exception:unknown method:" and names the method asked for. hivemind echoes params into the same -32602 assert ("Post a/x does not exist"), so a crafted permlink could mark a v2-capable node for the whole TTL. - Drop bridge.list_votes from the allowlist: it has no twin to fall back to, and a node lacking it would answer every call with an RPC error that neither fails over nor counts against the node. - v2 reads take object params only; positional ones answer like an unknown route, since the v1 fallback drops top_votes by name. - Note that an unmarked attempt makes two round trips once per node and method per TTL, and that the fallback attempt is timed on the v1 answer. - Tests: param echoes, per-node marks, fallback latency, positional params. Refs #119 --- dotnet/EcencyApi.Tests/SsrRpcTests.cs | 131 ++++++++++++++++-- dotnet/EcencyApi/Handlers/SsrRpc.cs | 13 +- .../EcencyApi/Infrastructure/HiveRpcClient.cs | 24 +++- 3 files changed, 149 insertions(+), 19 deletions(-) diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index ce9b4163..aa833507 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -27,6 +27,12 @@ private sealed class RpcStub : IAsyncDisposable public int DelayMs; public bool RpcError; public bool NullResult; + // Any other hivemind assert, e.g. one that echoes a param back. + public string? CustomError; + // Answer with this HTTP status instead of 200 (a node out of service). + public int Status = 200; + // Extra delay before an `unknown method` answer only. + public int UnknownDelayMs; public volatile string LastMethod = ""; // Methods this node answers `unknown method` for, the way a hivemind // without the v2 reads does (HTTP 200, code -32602, assert message). @@ -84,8 +90,9 @@ private async Task Loop() // param ("got an unexpected keyword argument"). var unexpectedArg = rawMethod.StartsWith("bridge.") && !rawMethod.EndsWith("_v2") && req?["params"] is JsonObject po && po.ContainsKey("top_votes"); - var assertion = unexpectedArg ? "got an unexpected keyword argument 'top_votes'" - : Unknown.Contains(rawMethod) ? "unknown method: bridge_api" + method : null; + var assertion = CustomError ?? (unexpectedArg ? "got an unexpected keyword argument 'top_votes'" + : Unknown.Contains(rawMethod) ? "unknown method: bridge_api" + method : null); + if (Unknown.Contains(rawMethod) && UnknownDelayMs > 0) await Task.Delay(UnknownDelayMs); var body = assertion is not null ? "{\"jsonrpc\":\"2.0\",\"id\":1,\"error\":{\"code\":-32602,\"message\":\"Assert Exception:" + assertion + "\",\"data\":{\"code\":10,\"extension\":{\"assertion_expression\":\"" + assertion + "\"}}}}" @@ -95,7 +102,7 @@ private async Task Loop() ? "{\"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.StatusCode = Status; ctx.Response.ContentType = "application/json"; ctx.Response.ContentLength64 = bytes.Length; try { await ctx.Response.OutputStream.WriteAsync(bytes); ctx.Response.Close(); } catch { } @@ -641,7 +648,7 @@ public void Allowlist_is_read_only_and_names_every_method_the_consumer_routes() "condenser_api.get_accounts", "condenser_api.get_content", "condenser_api.get_dynamic_global_properties", "condenser_api.get_trending_tags", "bridge.get_post_v2", "bridge.get_ranked_posts_v2", "bridge.get_account_posts_v2", - "bridge.get_discussion_v2", "bridge.list_votes", + "bridge.get_discussion_v2", }) { Assert.True(SsrRpc.Allowlist.ContainsKey(key), key); @@ -649,6 +656,10 @@ public void Allowlist_is_read_only_and_names_every_method_the_consumer_routes() } Assert.False(SsrRpc.Allowlist.ContainsKey("condenser_api.broadcast_transaction")); Assert.False(SsrRpc.Allowlist.ContainsKey("database_api.get_accounts")); + // Votes are read by the client-side votes dialog only, and a node at the + // head of the pool without list_votes would answer every call with an + // RPC error (no failover, counted as a fast success). + Assert.False(SsrRpc.Allowlist.ContainsKey("bridge.list_votes")); } [Fact] @@ -853,7 +864,7 @@ public async Task Once_the_mark_has_lapsed_v2_is_tried_first_again() } [Fact] - public async Task Stats_list_the_v2_methods_and_list_votes_with_their_class_and_the_node_marks() + public async Task Stats_list_the_v2_methods_with_their_class_and_the_node_marks() { await using var stub = new RpcStub { Unknown = { "bridge.get_ranked_posts_v2" } }; Use(stub); @@ -864,7 +875,6 @@ public async Task Stats_list_the_v2_methods_and_list_votes_with_their_class_and_ await SsrRpc.Resolve(Policy("bridge.get_ranked_posts_v2"), new JsonObject { ["sort"] = "trending", ["tag"] = "" }); await SsrRpc.Resolve(Policy("bridge.get_account_posts_v2"), new JsonObject { ["sort"] = "posts", ["account"] = "a" }); await SsrRpc.Resolve(Policy("bridge.get_discussion_v2"), P("a", "b")); - await SsrRpc.Resolve(Policy("bridge.list_votes"), new JsonObject { ["author"] = "a", ["permlink"] = "b", ["limit"] = 100 }); var stats = Request("GET", "/private-api/ssr/stats", "right-secret"); await SsrRpc.Stats(stats); @@ -874,7 +884,6 @@ public async Task Stats_list_the_v2_methods_and_list_votes_with_their_class_and_ Assert.Equal("heavy", methods["bridge.get_ranked_posts_v2"]?["class"]?.GetValue()); Assert.Equal("heavy", methods["bridge.get_account_posts_v2"]?["class"]?.GetValue()); Assert.Equal("heavy", methods["bridge.get_discussion_v2"]?["class"]?.GetValue()); - Assert.Equal("cheap", methods["bridge.list_votes"]?["class"]?.GetValue()); Assert.Equal(1, methods["bridge.get_ranked_posts_v2"]?["miss"]?.GetValue()); // Which node lacks which v2 read, so the fallback can be retired @@ -890,7 +899,7 @@ public async Task Stats_list_the_v2_methods_and_list_votes_with_their_class_and_ } [Fact] - public void V2_twins_take_the_v1_ttl_and_class_and_list_votes_is_a_cheap_point_read() + public void V2_twins_take_the_v1_ttl_and_class() { foreach (var v1 in new[] { "bridge.get_post", "bridge.get_ranked_posts", "bridge.get_account_posts", "bridge.get_discussion" }) { @@ -901,9 +910,107 @@ public void V2_twins_take_the_v1_ttl_and_class_and_list_votes_is_a_cheap_point_r Assert.Equal(one.Method, two.Fallback); Assert.Null(one.Fallback); } - var votes = Policy("bridge.list_votes"); - Assert.Equal(CallClass.Cheap, votes.Class); - Assert.Equal(Policy("bridge.get_post").TtlMs, votes.TtlMs); - Assert.Null(votes.Fallback); + } + + [Fact] + public async Task A_param_echoed_into_an_assert_or_another_methods_unknown_method_does_not_mark_the_node() + { + // hivemind echoes params into its -32602 asserts ("Post a/x does not + // exist", "invalid tag `x`"), so a crafted permlink must not be able to + // mark a v2-capable node as lacking v2. + await using var stub = new RpcStub { CustomError = "Post a/unknown method does not exist" }; + Use(stub); + var v2 = Policy("bridge.get_post_v2"); + Assert.Equal(SsrRpc.Outcome.RpcError, (await SsrRpc.Resolve(v2, PV2("a", "unknown method"))).Outcome); + // An echo that even ends with the method's name is still not the assert. + stub.CustomError = "Post a/unknown method: bridge_apiget_post_v2"; + Assert.Equal(SsrRpc.Outcome.RpcError, (await SsrRpc.Resolve(v2, PV2("a", "tag"))).Outcome); + // An unknown-method assert that names a different method is not this one. + stub.CustomError = "unknown method: bridge_apiget_ranked_posts_v2"; + Assert.Equal(SsrRpc.Outcome.RpcError, (await SsrRpc.Resolve(v2, PV2("a", "other"))).Outcome); + Assert.Equal(new[] { "bridge.get_post_v2", "bridge.get_post_v2", "bridge.get_post_v2" }, stub.Methods.ToArray()); + + stub.CustomError = null; + var ok = await SsrRpc.Resolve(v2, PV2("a", "real")); + Assert.StartsWith("{\"method\":\"get_post_v2\"", Encoding.UTF8.GetString(ok.Bytes)); + Assert.Empty(SsrRpc.Client.HealthSnapshot()[0]!["unsupported"]!.AsArray()); + } + + [Fact] + public async Task Positional_params_on_a_v2_read_answer_like_an_unknown_route() + { + // The v1 fallback drops `top_votes` by name; a positional one would reach + // a v1 node as one argument too many, so v2 reads take objects only. + await using var stub = new RpcStub(); + Use(stub); + SsrRpc.SecretDigest = SsrRpc.Digest("s"); + try + { + var arr = Request("POST", "/private-api/ssr/rpc", "s", + "{\"api\":\"bridge\",\"method\":\"get_post_v2\",\"params\":[\"a\",\"b\",\"\",0]}"); + await SsrRpc.Rpc(arr); + Assert.Equal(404, arr.Response.StatusCode); + Assert.Contains("Cannot POST /private-api/ssr/rpc", ResponseText(arr)); + Assert.Equal(0, stub.Hits); + + var obj = Request("POST", "/private-api/ssr/rpc", "s", + "{\"api\":\"bridge\",\"method\":\"get_post_v2\",\"params\":{\"author\":\"a\",\"permlink\":\"b\"}}"); + await SsrRpc.Rpc(obj); + Assert.Equal(200, obj.Response.StatusCode); + Assert.Equal(1, stub.Hits); + } + finally + { + SsrRpc.SecretDigest = null; + } + } + + [Fact] + public async Task The_mark_belongs_to_the_node_that_lacks_the_method_not_to_its_neighbours() + { + // The capable node is first in the pool; the lacking one is only reached + // while the capable one is out, and then the capable one is back. + await using var capable = new RpcStub { Status = 503 }; + await using var lacking = new RpcStub { Unknown = { "bridge.get_post_v2" } }; + Use(capable); + SsrRpc.Client = new HiveRpcClient(new[] { capable.Url, lacking.Url }, timeoutMs: 1000, failoverThreshold: 1); + var v2 = Policy("bridge.get_post_v2"); + + var first = await SsrRpc.Resolve(v2, PV2("a", "1")); + Assert.True(first.Outcome == SsrRpc.Outcome.Miss, $"{first.Outcome}: {first.Error}"); + Assert.Equal(new[] { "bridge.get_post_v2", "bridge.get_post" }, lacking.Methods.ToArray()); + + capable.Status = 200; + lacking.Status = 503; + var second = await SsrRpc.Resolve(v2, PV2("a", "2")); + Assert.True(second.Outcome == SsrRpc.Outcome.Miss, $"{second.Outcome}: {second.Error}"); + Assert.Equal("bridge.get_post_v2", capable.Methods.Last()); + Assert.StartsWith("{\"method\":\"get_post_v2\"", Encoding.UTF8.GetString(second.Bytes)); + + var snap = SsrRpc.Client.HealthSnapshot(); + Assert.Empty(snap[0]!["unsupported"]!.AsArray()); + Assert.Equal("bridge.get_post_v2", Assert.Single(snap[1]!["unsupported"]!.AsArray())!.GetValue()); + + // And the marked node, reached again, is asked v1 directly. + capable.Status = 503; + lacking.Status = 200; + var third = await SsrRpc.Resolve(v2, PV2("a", "3")); + Assert.True(third.Outcome == SsrRpc.Outcome.Miss, $"{third.Outcome}: {third.Error}"); + Assert.StartsWith("{\"method\":\"get_post\"", Encoding.UTF8.GetString(third.Bytes)); + Assert.Equal(1, lacking.Methods.Count(m => m == "bridge.get_post_v2")); + } + + [Fact] + public async Task A_fallback_attempt_records_the_latency_of_the_v1_answer_not_of_the_probe() + { + // The node is asked v1 directly for the next half hour, so the latency + // that ranks it is the v1 answer's; the one-off v2 probe is excluded. + await using var stub = new RpcStub { Unknown = { "bridge.get_post_v2" }, UnknownDelayMs = 400 }; + Use(stub); + var r = await SsrRpc.Resolve(Policy("bridge.get_post_v2"), PV2("a", "b")); + Assert.Equal(SsrRpc.Outcome.Miss, r.Outcome); + var node = SsrRpc.Client.HealthSnapshot()[0]!; + Assert.Equal(1, node["samples"]!.GetValue()); + Assert.True(node["ewma_ms"]!.GetValue() < 300, $"ewma {node["ewma_ms"]}"); } } diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index efed6dd3..eca406f6 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -58,13 +58,15 @@ internal sealed record MethodPolicy(string Api, string Method, int TtlMs, CallCl new MethodPolicy("bridge", "get_discussion", 30_000, CallClass.Heavy), new MethodPolicy("bridge", "get_post", 30_000, CallClass.Cheap), // Vote-free v2 twins: the v1 TTL and class, plus the v1 method a node - // without them is asked instead (see HiveRpcClient.CallMethod). A post's - // votes come separately from list_votes, a point read on the get_post TTL. + // without them is asked instead (see HiveRpcClient.CallMethod). + // bridge.list_votes is deliberately absent: only the client-side votes + // dialog reads it, and with no twin to fall back to, a node at the head of + // the pool that lacks it would answer every call with an RPC error, + // which neither fails over nor counts against the node. new MethodPolicy("bridge", "get_ranked_posts_v2", 15_000, CallClass.Heavy, Fallback: "get_ranked_posts"), new MethodPolicy("bridge", "get_account_posts_v2", 30_000, CallClass.Heavy, Fallback: "get_account_posts"), new MethodPolicy("bridge", "get_discussion_v2", 30_000, CallClass.Heavy, Fallback: "get_discussion"), new MethodPolicy("bridge", "get_post_v2", 30_000, CallClass.Cheap, Fallback: "get_post"), - new MethodPolicy("bridge", "list_votes", 30_000, CallClass.Cheap), new MethodPolicy("bridge", "get_profile", 60_000, CallClass.Cheap), new MethodPolicy("bridge", "get_profiles", 60_000, CallClass.Cheap), new MethodPolicy("bridge", "get_community", 300_000, CallClass.Cheap), @@ -449,7 +451,10 @@ public static async Task Rpc(HttpContext ctx) // 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)) + || !Allowlist.TryGetValue($"{api}.{method}", out var policy) + // A v2 read falls back by dropping `top_votes` by name, which a + // positional top_votes would survive; the consumer sends objects. + || (policy.Fallback is not null && @params is not JsonObject)) { await Routes.Fallback(ctx); return; diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index 978ced74..2b888b30 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs @@ -191,8 +191,21 @@ public RpcException(string message) : base(message) { } // hivemind without the method: HTTP 200, code -32602, "Assert // Exception:unknown method: bridge_apiget_post_v2" (verified 2026-11-12). - private static bool IsUnknownMethod(RpcException e) => - e.Message.Contains("unknown method", StringComparison.OrdinalIgnoreCase); + // The same code and prefix carry asserts that echo params ("Post a/x does + // not exist", "invalid tag `x`"), so the assert has to START with the + // unknown-method text and name the method asked for; otherwise one crafted + // permlink would mark a capable node for the whole TTL. + private const string UnknownMethodPrefix = "Assert Exception:unknown method:"; + + internal static bool IsUnknownMethod(string message, string qualifiedMethod) + { + if (!message.StartsWith(UnknownMethodPrefix, StringComparison.Ordinal)) return false; + var named = message[UnknownMethodPrefix.Length..].Trim(); + var bare = qualifiedMethod[(qualifiedMethod.LastIndexOf('.') + 1)..]; + // hivemind names it `_api`: bridge.get_post_v2 is + // bridge_apiget_post_v2. + return named.EndsWith("_api" + bare, StringComparison.Ordinal); + } private bool Unsupported(int nodeIndex, string method) { @@ -241,11 +254,16 @@ private IEnumerable UnsupportedMethods(int nodeIndex) => { result = await CallNode(node, useFallback ? fallbackBody! : body); } - catch (RpcException e) when (fallbackBody != null && !useFallback && IsUnknownMethod(e)) + catch (RpcException e) when (fallbackBody != null && !useFallback && IsUnknownMethod(e.Message, method)) { // The node is up, it just predates the method: same node, // older method, and remember it. Not a failure, not a failover. + // Two round trips against the caller's budget, but only once + // per (node, method) per TTL, and inside a fill the fill gate + // already bounds. _unsupportedUntil[(nodeIndex, method)] = _clock() + UnsupportedMethodTtlMs; + // Rank the node on the v1 answer it will give from now on, + // not on the probe that does not recur within the TTL. started = NowMs; result = await CallNode(node, fallbackBody); } From 33b1292dfb12b76b4a7f6fa460d5a6111669d6d4 Mon Sep 17 00:00:00 2001 From: feruzm Date: Thu, 12 Nov 2026 07:53:33 +0000 Subject: [PATCH 3/5] SSR read cache: one v2 probe per node, family-wide marks - Coalesce probes per (node, family). Until a node has a verdict, one call sends the v2 probe and concurrent calls for that node wait for its outcome (bounded by the probe's per-node timeout), then go v2 or v1 accordingly. A cold burst of N distinct v2 reads against a v1-only node now makes one v2 request plus N v1 requests instead of N of each. - A verdict either way holds for the TTL, so a v2-capable node is asked directly with no queueing; any answer other than unknown method, application errors included, proves the method. A probe that fails at the transport releases its waiters and the next call probes again. - The four v2 reads shipped together, so one unknown-method answer marks the whole family (bridge.*_v2); /stats lists the family per node. - Clone the fallback params, so one node may be passed as both params. - Time waiting for another call's probe is not node latency. - The test stub now serves requests concurrently, like a real node. Refs #119 --- dotnet/EcencyApi.Tests/SsrRpcTests.cs | 98 +++++++++++- dotnet/EcencyApi/Handlers/SsrRpc.cs | 6 +- .../EcencyApi/Infrastructure/HiveRpcClient.cs | 140 +++++++++++++----- 3 files changed, 208 insertions(+), 36 deletions(-) diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index aa833507..1647b91b 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -59,6 +59,14 @@ private async Task Loop() HttpListenerContext ctx; try { ctx = await _listener.GetContextAsync(); } catch { return; } + // Concurrently, like a real node: a burst must not queue here. + _ = Handle(ctx); + } + } + + private async Task Handle(HttpListenerContext ctx) + { + { var n = Interlocked.Increment(ref Hits); string reqBody; using (var reader = new StreamReader(ctx.Request.InputStream)) @@ -890,7 +898,7 @@ public async Task Stats_list_the_v2_methods_with_their_class_and_the_node_marks( // once the list stays empty. var node = Assert.Single(body["nodes"]!.AsArray())!; var unsupported = node["unsupported"]?.AsArray().Select(m => m!.GetValue()).ToArray(); - Assert.Equal(new[] { "bridge.get_ranked_posts_v2" }, unsupported); + Assert.Equal(new[] { SsrRpc.V2Family }, unsupported); } finally { @@ -989,7 +997,7 @@ public async Task The_mark_belongs_to_the_node_that_lacks_the_method_not_to_its_ var snap = SsrRpc.Client.HealthSnapshot(); Assert.Empty(snap[0]!["unsupported"]!.AsArray()); - Assert.Equal("bridge.get_post_v2", Assert.Single(snap[1]!["unsupported"]!.AsArray())!.GetValue()); + Assert.Equal(SsrRpc.V2Family, Assert.Single(snap[1]!["unsupported"]!.AsArray())!.GetValue()); // And the marked node, reached again, is asked v1 directly. capable.Status = 503; @@ -1013,4 +1021,90 @@ public async Task A_fallback_attempt_records_the_latency_of_the_v1_answer_not_of Assert.Equal(1, node["samples"]!.GetValue()); Assert.True(node["ewma_ms"]!.GetValue() < 300, $"ewma {node["ewma_ms"]}"); } + + [Fact] + public async Task A_cold_burst_of_v2_reads_against_a_v1_only_node_sends_one_probe() + { + // Distinct keys, so the cache's single-flight does not coalesce them; + // the probe is slow, so every fill reaches the node before it lands. + await using var stub = new RpcStub { Unknown = { "bridge.get_post_v2" }, UnknownDelayMs = 300 }; + Use(stub); + var v2 = Policy("bridge.get_post_v2"); + const int n = 8; + var results = await Task.WhenAll(Enumerable.Range(0, n).Select(i => SsrRpc.Resolve(v2, PV2("a", "p" + i)))); + Assert.All(results, r => Assert.True(r.Outcome == SsrRpc.Outcome.Miss, $"{r.Outcome}: {r.Error}")); + Assert.All(results, r => Assert.StartsWith("{\"method\":\"get_post\"", Encoding.UTF8.GetString(r.Bytes))); + Assert.Equal(1, stub.Methods.Count(m => m == "bridge.get_post_v2")); + Assert.Equal(n, stub.Methods.Count(m => m == "bridge.get_post")); + // Time spent waiting for another call's probe is not this node's latency. + var node = SsrRpc.Client.HealthSnapshot()[0]!; + Assert.Equal(n, node["samples"]!.GetValue()); + Assert.True(node["ewma_ms"]!.GetValue() < 150, $"ewma {node["ewma_ms"]}"); + } + + [Fact] + public async Task A_cold_burst_on_a_v2_node_waits_for_one_probe_and_then_runs_in_parallel() + { + // Waiters go v2 once the probe proves the family; after that the node + // is known capable and reads do not queue behind one another. + await using var stub = new RpcStub { DelayMs = 200 }; + Use(stub); + var v2 = Policy("bridge.get_post_v2"); + const int n = 6; + var first = await Task.WhenAll(Enumerable.Range(0, n).Select(i => SsrRpc.Resolve(v2, PV2("a", "p" + i)))); + Assert.All(first, r => Assert.True(r.Outcome == SsrRpc.Outcome.Miss, $"{r.Outcome}: {r.Error}")); + Assert.Equal(n, stub.Methods.Count(m => m == "bridge.get_post_v2")); + Assert.Equal(0, stub.Methods.Count(m => m == "bridge.get_post")); + + var sw = System.Diagnostics.Stopwatch.StartNew(); + var second = await Task.WhenAll(Enumerable.Range(0, n).Select(i => SsrRpc.Resolve(v2, PV2("b", "p" + i)))); + sw.Stop(); + Assert.All(second, r => Assert.Equal(SsrRpc.Outcome.Miss, r.Outcome)); + Assert.True(sw.ElapsedMilliseconds < 2 * 200 + 300, $"took {sw.ElapsedMilliseconds} ms"); + } + + [Fact] + public async Task One_unknown_method_answer_marks_all_four_v2_reads_on_that_node() + { + await using var stub = new RpcStub + { + Unknown = { "bridge.get_post_v2", "bridge.get_ranked_posts_v2", "bridge.get_account_posts_v2", "bridge.get_discussion_v2" }, + }; + Use(stub); + await SsrRpc.Resolve(Policy("bridge.get_post_v2"), PV2("a", "b")); + await SsrRpc.Resolve(Policy("bridge.get_ranked_posts_v2"), new JsonObject { ["sort"] = "trending", ["tag"] = "" }); + await SsrRpc.Resolve(Policy("bridge.get_account_posts_v2"), new JsonObject { ["sort"] = "posts", ["account"] = "a" }); + await SsrRpc.Resolve(Policy("bridge.get_discussion_v2"), P("a", "b")); + Assert.Equal( + new[] { "bridge.get_post_v2", "bridge.get_post", "bridge.get_ranked_posts", "bridge.get_account_posts", "bridge.get_discussion" }, + stub.Methods.ToArray()); + } + + [Fact] + public async Task The_same_node_may_serve_as_params_and_as_fallback_params() + { + await using var stub = new RpcStub { Unknown = { "bridge.get_post_v2" } }; + var client = new HiveRpcClient(new[] { stub.Url }, timeoutMs: 1000, failoverThreshold: 1); + var p = P("a", "b"); + var result = await client.CallMethod("bridge.get_post_v2", p, fallback: ("bridge.get_post", p, "bridge.*_v2")); + Assert.Equal("get_post", result?["method"]?.GetValue()); + } + + [Fact] + public async Task An_rpc_error_from_the_probe_proves_the_method_so_a_burst_of_errors_runs_in_parallel() + { + // A missing post answers with an assert, which only a node that has the + // method can give. If that taught nothing, every waiter would probe in + // turn and the burst would queue one round trip at a time. + await using var stub = new RpcStub { DelayMs = 200, RpcError = true }; + Use(stub); + var v2 = Policy("bridge.get_post_v2"); + const int n = 6; + var sw = System.Diagnostics.Stopwatch.StartNew(); + var results = await Task.WhenAll(Enumerable.Range(0, n).Select(i => SsrRpc.Resolve(v2, PV2("a", "p" + i)))); + sw.Stop(); + Assert.All(results, r => Assert.Equal(SsrRpc.Outcome.RpcError, r.Outcome)); + Assert.Equal(n, stub.Methods.Count(m => m == "bridge.get_post_v2")); + Assert.True(sw.ElapsedMilliseconds < 2 * 200 + 300, $"took {sw.ElapsedMilliseconds} ms"); + } } diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index eca406f6..43bf91c5 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -335,6 +335,10 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa // The answer is cached under the requested (v2) key as returned, in either // shape: the route stays a raw pass-through and the consumer's normalizer // takes both, so one fill per key still serves every reader of that key. + // The four v2 reads shipped together in one hivemind release, so a node + // lacking one lacks them all: one mark (and one probe) covers the family. + internal const string V2Family = "bridge.*_v2"; + internal static JsonNode V1Params(JsonNode @params) { var copy = @params.DeepClone(); @@ -400,7 +404,7 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key // re-parented into the envelope, so it travels as a clone. result = await Client.CallMethod($"{policy.Api}.{policy.Method}", @params.DeepClone(), callClass: CallClasses ? policy.Class : CallClass.Cheap, - fallback: policy.Fallback is { } v1 ? ($"{policy.Api}.{v1}", V1Params(@params)) : null); + fallback: policy.Fallback is { } v1 ? ($"{policy.Api}.{v1}", V1Params(@params), V2Family) : null); } finally { diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index 2b888b30..70140ac2 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs @@ -15,7 +15,8 @@ namespace EcencyApi.Infrastructure; /// - Overload statuses (429/502/503/504) advance to the next node immediately /// instead of burning a same-node retry. /// - An `unknown method` answer to a call that names a fallback method asks -/// the same node the fallback, and marks the node for a while (CallMethod). +/// the same node the fallback, and marks the node and the method's family +/// for a while; one call probes a node at a time (CallMethod). /// /// Latency is tracked per call class (see ) because the /// reads made through this client are bimodal in cost. Failure state is not: it @@ -33,9 +34,15 @@ public sealed class HiveRpcClient private readonly Func _clock; private int _seq; - // Per node, the methods it answered `unknown method` for and until when (on - // _clock). Bounded by nodes x fallback-capable methods; see CallMethod. - private readonly ConcurrentDictionary<(int Node, string Method), long> _unsupportedUntil = new(); + // Per node and fallback family, what is known about the node serving the + // family: nothing (no entry), a probe in flight, or a verdict until a time + // on _clock. Bounded by nodes x families; see CallMethod. + private readonly ConcurrentDictionary<(int Node, string Family), Capability> _capability = new(); + + /// null while is in flight. + /// Completes with the probe's verdict, or null when the + /// probe learned nothing (the node failed), after the entry is updated. + private sealed record Capability(bool? Supported, long Until, Task? Probe); // timeoutMs 2000 / failoverThreshold 2 mirror the dhive Client options the // Node service constructed its clients with. @@ -91,9 +98,9 @@ public JsonArray HealthSnapshot() node["rate_limited_for_ms"] = v.RateLimitedForMs; node["parked_for_ms"] = v.FailureParkedForMs; node["failure_rate"] = Math.Round(v.FailureRate, 3); - // Methods this node is asked the fallback for until its mark lapses. + // Families this node is asked the fallback for until its mark lapses. var unsupported = new JsonArray(); - foreach (var m in UnsupportedMethods(v.Index)) unsupported.Add(m); + foreach (var m in UnsupportedFamilies(v.Index)) unsupported.Add(m); node["unsupported"] = unsupported; arr.Add(node); } @@ -111,11 +118,12 @@ public RpcException(string message) : base(message) { } /// preference before the first well-formed answer is accepted as-is. private const int MaxPreferenceProbes = 2; - /// How long a node that answered `unknown method` is asked the - /// fallback instead. Long enough that a node lacking the method costs one - /// wasted round trip per method per half hour, short enough that a node - /// upgraded in place is used for the new method again within the half hour, - /// with no restart. + /// How long a verdict on a node and family holds: a node that + /// answered `unknown method` is asked the fallback, one that answered anything + /// else is asked directly, with no probe either way. Long enough that a node + /// lacking the family costs one wasted round trip per half hour, short enough + /// that a node upgraded in place is used for the new methods again within the + /// half hour, with no restart. internal const long UnsupportedMethodTtlMs = 30 * 60 * 1000; // ---- calls ------------------------------------------------------------- @@ -166,19 +174,26 @@ public RpcException(string message) : base(message) { } /// for every hivemind (`bridge`) read; works for condenser_api too. /// /// An older method serving the same read, for nodes - /// that do not have this one yet. A node that answers `unknown method` is not - /// failed over: it is asked the fallback at once, in the same attempt, and is - /// asked the fallback directly for . - /// Whatever that node answers is returned as is. + /// that do not have this one yet, and the family of new methods that ship + /// together (one verdict covers them all). A node that answers `unknown + /// method` is not failed over: it is asked the fallback at once, in the same + /// attempt, and is asked the fallback directly for + /// . A node that answers anything else is + /// known to serve the family for as long. Until a node has a verdict, one call + /// probes it and the others wait for that verdict instead of each probing. + /// Whatever the node answers is returned as is. public Task CallMethod(string qualifiedMethod, JsonNode @params, Func? validateResult = null, Func? preferResult = null, CallClass callClass = CallClass.Cheap, - (string Method, JsonNode Params)? fallback = null) + (string Method, JsonNode Params, string Family)? fallback = null) { var request = Envelope(qualifiedMethod, @params); - var fallbackRequest = fallback is { } f ? Envelope(f.Method, f.Params) : null; - return Send(request, qualifiedMethod, validateResult, preferResult, callClass, fallbackRequest); + // A clone: the caller may pass one node as both params, and a node can + // hang off only one envelope. + var fallbackRequest = fallback is { } f ? Envelope(f.Method, f.Params.DeepClone()) : null; + return Send(request, qualifiedMethod, validateResult, preferResult, callClass, fallbackRequest, + fallback?.Family); } private JsonObject Envelope(string qualifiedMethod, JsonNode @params) => new() @@ -207,25 +222,66 @@ internal static bool IsUnknownMethod(string message, string qualifiedMethod) return named.EndsWith("_api" + bare, StringComparison.Ordinal); } - private bool Unsupported(int nodeIndex, string method) + private IEnumerable UnsupportedFamilies(int nodeIndex) { - if (!_unsupportedUntil.TryGetValue((nodeIndex, method), out var until)) return false; - if (_clock() < until) return true; - _unsupportedUntil.TryRemove(new KeyValuePair<(int, string), long>((nodeIndex, method), until)); - return false; - } - - private IEnumerable UnsupportedMethods(int nodeIndex) => - _unsupportedUntil.Keys.Where(k => k.Node == nodeIndex && Unsupported(nodeIndex, k.Method)) - .Select(k => k.Method) + var now = _clock(); + return _capability + .Where(kv => kv.Key.Node == nodeIndex && kv.Value.Supported == false && now < kv.Value.Until) + .Select(kv => kv.Key.Family) .OrderBy(m => m, StringComparer.Ordinal) .ToArray(); + } + + /// How to ask this node for a method of this family: the fallback + /// (true), the new method (false), and whether this call is the probe whose + /// verdict the caller must then report through . While + /// another call probes the node, this waits for that verdict; the wait is + /// bounded by the probe's own per-node timeout. + private async Task<(bool UseFallback, TaskCompletionSource? Probe)> Route(int nodeIndex, string family) + { + var key = (nodeIndex, family); + while (true) + { + _capability.TryGetValue(key, out var cap); + if (cap?.Probe is { } inFlight) + { + await inFlight; + continue; + } + if (cap is not null && _clock() < cap.Until) return (cap.Supported == false, null); + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var probing = new Capability(null, 0, tcs.Task); + var won = cap is null ? _capability.TryAdd(key, probing) : _capability.TryUpdate(key, probing, cap); + if (won) return (false, tcs); + } + } + + /// Record a verdict for a node and family. A probe settles its own + /// entry (null: it learned nothing, so the next caller probes again) and then + /// releases its waiters; a call outside a probe overwrites any entry. + private void Settle(int nodeIndex, string family, bool? supported, TaskCompletionSource? probe) + { + var key = (nodeIndex, family); + var verdict = supported is { } v ? new Capability(v, _clock() + UnsupportedMethodTtlMs, null) : null; + if (probe is null) + { + if (verdict is not null) _capability[key] = verdict; + return; + } + if (_capability.TryGetValue(key, out var cap) && ReferenceEquals(cap.Probe, probe.Task)) + { + if (verdict is null) _capability.TryRemove(new KeyValuePair<(int, string), Capability>(key, cap)); + else _capability.TryUpdate(key, verdict, cap); + } + probe.TrySetResult(supported); + } private async Task Send(JsonObject request, string method, Func? validateResult, Func? preferResult, CallClass callClass, - JsonObject? fallbackRequest = null) + JsonObject? fallbackRequest = null, + string? family = null) { // JsJson: a lone-surrogate username from a client token must serialize // (JSON.stringify semantics) instead of throwing in the writer. @@ -246,27 +302,42 @@ private IEnumerable UnsupportedMethods(int nodeIndex) => // The ordering above is a snapshot; admission is decided now. if (!_health.TryBeginAttempt(nodeIndex)) break; var started = NowMs; + TaskCompletionSource? probe = null; try { - var useFallback = fallbackBody != null && Unsupported(nodeIndex, method); + var useFallback = false; + if (fallbackBody != null) + { + (useFallback, probe) = await Route(nodeIndex, family ?? method); + started = NowMs; // the wait for another call's probe is not this node's latency + } JsonNode? result; try { result = await CallNode(node, useFallback ? fallbackBody! : body); + if (probe != null) { Settle(nodeIndex, family ?? method, true, probe); probe = null; } } catch (RpcException e) when (fallbackBody != null && !useFallback && IsUnknownMethod(e.Message, method)) { // The node is up, it just predates the method: same node, // older method, and remember it. Not a failure, not a failover. // Two round trips against the caller's budget, but only once - // per (node, method) per TTL, and inside a fill the fill gate - // already bounds. - _unsupportedUntil[(nodeIndex, method)] = _clock() + UnsupportedMethodTtlMs; + // per (node, family) per TTL: concurrent calls wait for this + // verdict instead of probing too. + Settle(nodeIndex, family ?? method, false, probe); + probe = null; // Rank the node on the v1 answer it will give from now on, // not on the probe that does not recur within the TTL. started = NowMs; result = await CallNode(node, fallbackBody); } + catch (RpcException) when (probe != null) + { + // Any other application error proves the method exists. + Settle(nodeIndex, family ?? method, true, probe); + probe = null; + throw; + } if (validateResult != null && !validateResult(result)) { // Don't burn a same-node retry: a node serving malformed @@ -332,6 +403,9 @@ private IEnumerable UnsupportedMethods(int nodeIndex) => } finally { + // A probe that ended in a transport failure learned nothing: + // release its waiters so one of them probes again. + if (probe != null) Settle(nodeIndex, family ?? method, null, probe); _health.EndAttempt(nodeIndex); } } From 3c0ca3582193f608c8f9d2fc60bddeb1e54bf8bb Mon Sep 17 00:00:00 2001 From: feruzm Date: Thu, 12 Nov 2026 08:03:49 +0000 Subject: [PATCH 4/5] SSR read cache: drop v2 probe coalescing, keep family-wide marks Waiting for one in-flight probe per node coupled readers to that probe's fate: a probe that hung at the transport turned a burst into a queue of node timeouts, and on v2-capable nodes readers queued behind the probe at cold start and at every TTL expiry. An unknown-method answer comes back in one fast round trip, so letting each call decide on its own is cheap: calls that reach an unmarked node together may each ask v2 once per TTL, then the mark sends them to v1. - Every call sends v1 if the node is marked lacking the family, else v2, with the same-node v1 retry on an exact unknown-method answer. No in-flight probe state, no waiting, no "supported" verdict; only the lacks mark expires (30 min). - Keep family-wide marks (bridge.*_v2) listed per node on /stats, the clone of the fallback params, and the concurrent test stub. - Specs: a cold burst against a v1-only node finishes in one probe plus v1 per call; a burst with a hanging first node fails over in parallel; a burst on a v2 node runs in parallel cold and after a TTL advance. Refs #119 --- dotnet/EcencyApi.Tests/SsrRpcTests.cs | 91 ++++++++------ dotnet/EcencyApi/Handlers/SsrRpc.cs | 2 +- .../EcencyApi/Infrastructure/HiveRpcClient.cs | 115 ++++-------------- 3 files changed, 83 insertions(+), 125 deletions(-) diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 1647b91b..9351a0cc 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -1023,46 +1023,84 @@ public async Task A_fallback_attempt_records_the_latency_of_the_v1_answer_not_of } [Fact] - public async Task A_cold_burst_of_v2_reads_against_a_v1_only_node_sends_one_probe() + public async Task A_cold_burst_against_a_v1_only_node_completes_in_one_probe_plus_v1_each() { - // Distinct keys, so the cache's single-flight does not coalesce them; - // the probe is slow, so every fill reaches the node before it lands. + // Distinct keys, so the cache's single-flight does not coalesce them, + // and the probe is slow, so every call reaches the node before any mark + // exists. Each may probe (up to N), but none waits for another. await using var stub = new RpcStub { Unknown = { "bridge.get_post_v2" }, UnknownDelayMs = 300 }; Use(stub); var v2 = Policy("bridge.get_post_v2"); const int n = 8; + var sw = System.Diagnostics.Stopwatch.StartNew(); var results = await Task.WhenAll(Enumerable.Range(0, n).Select(i => SsrRpc.Resolve(v2, PV2("a", "p" + i)))); + sw.Stop(); Assert.All(results, r => Assert.True(r.Outcome == SsrRpc.Outcome.Miss, $"{r.Outcome}: {r.Error}")); Assert.All(results, r => Assert.StartsWith("{\"method\":\"get_post\"", Encoding.UTF8.GetString(r.Bytes))); - Assert.Equal(1, stub.Methods.Count(m => m == "bridge.get_post_v2")); + Assert.InRange(stub.Methods.Count(m => m == "bridge.get_post_v2"), 1, n); Assert.Equal(n, stub.Methods.Count(m => m == "bridge.get_post")); - // Time spent waiting for another call's probe is not this node's latency. + Assert.True(sw.ElapsedMilliseconds < 300 + 250, $"took {sw.ElapsedMilliseconds} ms"); + // Each sample is the v1 answer, not the probe before it. var node = SsrRpc.Client.HealthSnapshot()[0]!; Assert.Equal(n, node["samples"]!.GetValue()); Assert.True(node["ewma_ms"]!.GetValue() < 150, $"ewma {node["ewma_ms"]}"); } [Fact] - public async Task A_cold_burst_on_a_v2_node_waits_for_one_probe_and_then_runs_in_parallel() + public async Task A_burst_with_a_hanging_first_node_fails_over_in_parallel_not_in_a_queue() { - // Waiters go v2 once the probe proves the family; after that the node - // is known capable and reads do not queue behind one another. - await using var stub = new RpcStub { DelayMs = 200 }; + // Regression: when calls waited for one probe, a probe that hung made + // waiter k wait k node timeouts. + await using var hanging = new RpcStub { DelayMs = 5_000 }; + await using var healthy = new RpcStub(); + Use(hanging); + SsrRpc.Client = new HiveRpcClient(new[] { hanging.Url, healthy.Url }, timeoutMs: 300, failoverThreshold: 1); + var v2 = Policy("bridge.get_post_v2"); + const int n = 8; + var sw = System.Diagnostics.Stopwatch.StartNew(); + var results = await Task.WhenAll(Enumerable.Range(0, n).Select(async i => + { + var t = System.Diagnostics.Stopwatch.StartNew(); + var r = await SsrRpc.Resolve(v2, PV2("a", "p" + i)); + return (r, t.ElapsedMilliseconds); + })); + sw.Stop(); + Assert.All(results, x => Assert.True(x.r.Outcome == SsrRpc.Outcome.Miss, $"{x.r.Outcome}: {x.r.Error}")); + Assert.All(results, x => Assert.True(x.ElapsedMilliseconds < 2 * 300, $"one call took {x.ElapsedMilliseconds} ms")); + Assert.Equal(n, healthy.Methods.Count(m => m == "bridge.get_post_v2")); + } + + [Fact] + public async Task A_burst_on_a_v2_node_runs_in_parallel_cold_and_after_the_ttl() + { + await using var stub = new RpcStub { DelayMs = 400 }; Use(stub); + long now = 1_000_000; + SsrRpc.Client = new HiveRpcClient(new[] { stub.Url }, 1000, 1, () => now); var v2 = Policy("bridge.get_post_v2"); const int n = 6; - var first = await Task.WhenAll(Enumerable.Range(0, n).Select(i => SsrRpc.Resolve(v2, PV2("a", "p" + i)))); - Assert.All(first, r => Assert.True(r.Outcome == SsrRpc.Outcome.Miss, $"{r.Outcome}: {r.Error}")); - Assert.Equal(n, stub.Methods.Count(m => m == "bridge.get_post_v2")); - Assert.Equal(0, stub.Methods.Count(m => m == "bridge.get_post")); - var sw = System.Diagnostics.Stopwatch.StartNew(); - var second = await Task.WhenAll(Enumerable.Range(0, n).Select(i => SsrRpc.Resolve(v2, PV2("b", "p" + i)))); - sw.Stop(); - Assert.All(second, r => Assert.Equal(SsrRpc.Outcome.Miss, r.Outcome)); - Assert.True(sw.ElapsedMilliseconds < 2 * 200 + 300, $"took {sw.ElapsedMilliseconds} ms"); + async Task Burst(string author) + { + var results = await Task.WhenAll(Enumerable.Range(0, n).Select(async i => + { + var t = System.Diagnostics.Stopwatch.StartNew(); + var r = await SsrRpc.Resolve(v2, PV2(author, "p" + i)); + return (r, t.ElapsedMilliseconds); + })); + // One round trip each: none of them waited for another's answer. + Assert.All(results, x => Assert.True(x.r.Outcome == SsrRpc.Outcome.Miss, $"{x.r.Outcome}: {x.r.Error}")); + Assert.All(results, x => Assert.True(x.ElapsedMilliseconds < 400 + 200, $"one call took {x.ElapsedMilliseconds} ms")); + } + + await Burst("cold"); + now += HiveRpcClient.UnsupportedMethodTtlMs + 1; + await Burst("later"); + Assert.Equal(2 * n, stub.Methods.Count(m => m == "bridge.get_post_v2")); + Assert.Equal(0, stub.Methods.Count(m => m == "bridge.get_post")); } + [Fact] public async Task One_unknown_method_answer_marks_all_four_v2_reads_on_that_node() { @@ -1090,21 +1128,4 @@ public async Task The_same_node_may_serve_as_params_and_as_fallback_params() Assert.Equal("get_post", result?["method"]?.GetValue()); } - [Fact] - public async Task An_rpc_error_from_the_probe_proves_the_method_so_a_burst_of_errors_runs_in_parallel() - { - // A missing post answers with an assert, which only a node that has the - // method can give. If that taught nothing, every waiter would probe in - // turn and the burst would queue one round trip at a time. - await using var stub = new RpcStub { DelayMs = 200, RpcError = true }; - Use(stub); - var v2 = Policy("bridge.get_post_v2"); - const int n = 6; - var sw = System.Diagnostics.Stopwatch.StartNew(); - var results = await Task.WhenAll(Enumerable.Range(0, n).Select(i => SsrRpc.Resolve(v2, PV2("a", "p" + i)))); - sw.Stop(); - Assert.All(results, r => Assert.Equal(SsrRpc.Outcome.RpcError, r.Outcome)); - Assert.Equal(n, stub.Methods.Count(m => m == "bridge.get_post_v2")); - Assert.True(sw.ElapsedMilliseconds < 2 * 200 + 300, $"took {sw.ElapsedMilliseconds} ms"); - } } diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index 43bf91c5..495098e0 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -336,7 +336,7 @@ internal static async Task Resolve(MethodPolicy policy, JsonNode @pa // shape: the route stays a raw pass-through and the consumer's normalizer // takes both, so one fill per key still serves every reader of that key. // The four v2 reads shipped together in one hivemind release, so a node - // lacking one lacks them all: one mark (and one probe) covers the family. + // lacking one lacks them all: one mark covers the family. internal const string V2Family = "bridge.*_v2"; internal static JsonNode V1Params(JsonNode @params) diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index 70140ac2..f5b9e28d 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs @@ -16,7 +16,7 @@ namespace EcencyApi.Infrastructure; /// instead of burning a same-node retry. /// - An `unknown method` answer to a call that names a fallback method asks /// the same node the fallback, and marks the node and the method's family -/// for a while; one call probes a node at a time (CallMethod). +/// for a while (CallMethod). /// /// Latency is tracked per call class (see ) because the /// reads made through this client are bimodal in cost. Failure state is not: it @@ -34,15 +34,9 @@ public sealed class HiveRpcClient private readonly Func _clock; private int _seq; - // Per node and fallback family, what is known about the node serving the - // family: nothing (no entry), a probe in flight, or a verdict until a time - // on _clock. Bounded by nodes x families; see CallMethod. - private readonly ConcurrentDictionary<(int Node, string Family), Capability> _capability = new(); - - /// null while is in flight. - /// Completes with the probe's verdict, or null when the - /// probe learned nothing (the node failed), after the entry is updated. - private sealed record Capability(bool? Supported, long Until, Task? Probe); + // Per node, the fallback families it answered `unknown method` for and until + // when (on _clock). Bounded by nodes x families; see CallMethod. + private readonly ConcurrentDictionary<(int Node, string Family), long> _lacksUntil = new(); // timeoutMs 2000 / failoverThreshold 2 mirror the dhive Client options the // Node service constructed its clients with. @@ -118,10 +112,9 @@ public RpcException(string message) : base(message) { } /// preference before the first well-formed answer is accepted as-is. private const int MaxPreferenceProbes = 2; - /// How long a verdict on a node and family holds: a node that - /// answered `unknown method` is asked the fallback, one that answered anything - /// else is asked directly, with no probe either way. Long enough that a node - /// lacking the family costs one wasted round trip per half hour, short enough + /// How long a node that answered `unknown method` is asked the + /// fallback for the whole family. Long enough that a node lacking the family + /// costs a cold round of wasted round trips once per half hour, short enough /// that a node upgraded in place is used for the new methods again within the /// half hour, with no restart. internal const long UnsupportedMethodTtlMs = 30 * 60 * 1000; @@ -178,9 +171,11 @@ public RpcException(string message) : base(message) { } /// together (one verdict covers them all). A node that answers `unknown /// method` is not failed over: it is asked the fallback at once, in the same /// attempt, and is asked the fallback directly for - /// . A node that answers anything else is - /// known to serve the family for as long. Until a node has a verdict, one call - /// probes it and the others wait for that verdict instead of each probing. + /// . Each call decides on its own and + /// never waits on another: calls that reach an unmarked node together each + /// ask the new method, and an `unknown method` comes back in one fast round + /// trip, so a cold round of those per node per TTL is cheap, while tying + /// readers to one probe's fate turned a hung probe into a queue. /// Whatever the node answers is returned as is. public Task CallMethod(string qualifiedMethod, JsonNode @params, Func? validateResult = null, @@ -222,59 +217,19 @@ internal static bool IsUnknownMethod(string message, string qualifiedMethod) return named.EndsWith("_api" + bare, StringComparison.Ordinal); } - private IEnumerable UnsupportedFamilies(int nodeIndex) + private bool Lacks(int nodeIndex, string family) { - var now = _clock(); - return _capability - .Where(kv => kv.Key.Node == nodeIndex && kv.Value.Supported == false && now < kv.Value.Until) - .Select(kv => kv.Key.Family) - .OrderBy(m => m, StringComparer.Ordinal) - .ToArray(); + if (!_lacksUntil.TryGetValue((nodeIndex, family), out var until)) return false; + if (_clock() < until) return true; + _lacksUntil.TryRemove(new KeyValuePair<(int, string), long>((nodeIndex, family), until)); + return false; } - /// How to ask this node for a method of this family: the fallback - /// (true), the new method (false), and whether this call is the probe whose - /// verdict the caller must then report through . While - /// another call probes the node, this waits for that verdict; the wait is - /// bounded by the probe's own per-node timeout. - private async Task<(bool UseFallback, TaskCompletionSource? Probe)> Route(int nodeIndex, string family) - { - var key = (nodeIndex, family); - while (true) - { - _capability.TryGetValue(key, out var cap); - if (cap?.Probe is { } inFlight) - { - await inFlight; - continue; - } - if (cap is not null && _clock() < cap.Until) return (cap.Supported == false, null); - var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - var probing = new Capability(null, 0, tcs.Task); - var won = cap is null ? _capability.TryAdd(key, probing) : _capability.TryUpdate(key, probing, cap); - if (won) return (false, tcs); - } - } - - /// Record a verdict for a node and family. A probe settles its own - /// entry (null: it learned nothing, so the next caller probes again) and then - /// releases its waiters; a call outside a probe overwrites any entry. - private void Settle(int nodeIndex, string family, bool? supported, TaskCompletionSource? probe) - { - var key = (nodeIndex, family); - var verdict = supported is { } v ? new Capability(v, _clock() + UnsupportedMethodTtlMs, null) : null; - if (probe is null) - { - if (verdict is not null) _capability[key] = verdict; - return; - } - if (_capability.TryGetValue(key, out var cap) && ReferenceEquals(cap.Probe, probe.Task)) - { - if (verdict is null) _capability.TryRemove(new KeyValuePair<(int, string), Capability>(key, cap)); - else _capability.TryUpdate(key, verdict, cap); - } - probe.TrySetResult(supported); - } + private IEnumerable UnsupportedFamilies(int nodeIndex) => + _lacksUntil.Keys.Where(k => k.Node == nodeIndex && Lacks(nodeIndex, k.Family)) + .Select(k => k.Family) + .OrderBy(m => m, StringComparer.Ordinal) + .ToArray(); private async Task Send(JsonObject request, string method, Func? validateResult, @@ -302,42 +257,27 @@ private void Settle(int nodeIndex, string family, bool? supported, TaskCompletio // The ordering above is a snapshot; admission is decided now. if (!_health.TryBeginAttempt(nodeIndex)) break; var started = NowMs; - TaskCompletionSource? probe = null; try { - var useFallback = false; - if (fallbackBody != null) - { - (useFallback, probe) = await Route(nodeIndex, family ?? method); - started = NowMs; // the wait for another call's probe is not this node's latency - } + var useFallback = fallbackBody != null && Lacks(nodeIndex, family ?? method); JsonNode? result; try { result = await CallNode(node, useFallback ? fallbackBody! : body); - if (probe != null) { Settle(nodeIndex, family ?? method, true, probe); probe = null; } } catch (RpcException e) when (fallbackBody != null && !useFallback && IsUnknownMethod(e.Message, method)) { // The node is up, it just predates the method: same node, // older method, and remember it. Not a failure, not a failover. // Two round trips against the caller's budget, but only once - // per (node, family) per TTL: concurrent calls wait for this - // verdict instead of probing too. - Settle(nodeIndex, family ?? method, false, probe); - probe = null; + // per (node, family) per TTL for each call that reached the + // node before the mark, and inside a fill the gate bounds. + _lacksUntil[(nodeIndex, family ?? method)] = _clock() + UnsupportedMethodTtlMs; // Rank the node on the v1 answer it will give from now on, // not on the probe that does not recur within the TTL. started = NowMs; result = await CallNode(node, fallbackBody); } - catch (RpcException) when (probe != null) - { - // Any other application error proves the method exists. - Settle(nodeIndex, family ?? method, true, probe); - probe = null; - throw; - } if (validateResult != null && !validateResult(result)) { // Don't burn a same-node retry: a node serving malformed @@ -403,9 +343,6 @@ private void Settle(int nodeIndex, string family, bool? supported, TaskCompletio } finally { - // A probe that ended in a transport failure learned nothing: - // release its waiters so one of them probes again. - if (probe != null) Settle(nodeIndex, family ?? method, null, probe); _health.EndAttempt(nodeIndex); } } From d541bc177f62c57941bd6149ec6d1eb775df4bfe Mon Sep 17 00:00:00 2001 From: feruzm Date: Thu, 12 Nov 2026 08:06:59 +0000 Subject: [PATCH 5/5] SSR read cache: widen v2 timing spec margins, keep stub handlers quiet after close The burst specs now bound what a queue regression would cost instead of sitting a few hundred ms above the happy path, so they hold on a loaded host. Stub responses are written inside the try, so a handler that wakes after the listener closed no longer raises an unobserved exception. The family key is always set when a fallback is, so the unused per-method default is gone. Refs #119 --- dotnet/EcencyApi.Tests/SsrRpcTests.cs | 24 ++++++++++++------- .../EcencyApi/Infrastructure/HiveRpcClient.cs | 4 ++-- 2 files changed, 18 insertions(+), 10 deletions(-) diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 9351a0cc..17e096dd 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -110,10 +110,17 @@ private async Task Handle(HttpListenerContext ctx) ? "{\"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 = Status; - ctx.Response.ContentType = "application/json"; - ctx.Response.ContentLength64 = bytes.Length; - try { await ctx.Response.OutputStream.WriteAsync(bytes); ctx.Response.Close(); } catch { } + // Inside the try: a handler that wakes after the listener closed + // (the hanging-node spec) must not throw an unobserved exception. + try + { + ctx.Response.StatusCode = Status; + ctx.Response.ContentType = "application/json"; + ctx.Response.ContentLength64 = bytes.Length; + await ctx.Response.OutputStream.WriteAsync(bytes); + ctx.Response.Close(); + } + catch { } } } @@ -1039,11 +1046,11 @@ public async Task A_cold_burst_against_a_v1_only_node_completes_in_one_probe_plu Assert.All(results, r => Assert.StartsWith("{\"method\":\"get_post\"", Encoding.UTF8.GetString(r.Bytes))); Assert.InRange(stub.Methods.Count(m => m == "bridge.get_post_v2"), 1, n); Assert.Equal(n, stub.Methods.Count(m => m == "bridge.get_post")); - Assert.True(sw.ElapsedMilliseconds < 300 + 250, $"took {sw.ElapsedMilliseconds} ms"); + Assert.True(sw.ElapsedMilliseconds < 1000, $"took {sw.ElapsedMilliseconds} ms"); // Each sample is the v1 answer, not the probe before it. var node = SsrRpc.Client.HealthSnapshot()[0]!; Assert.Equal(n, node["samples"]!.GetValue()); - Assert.True(node["ewma_ms"]!.GetValue() < 150, $"ewma {node["ewma_ms"]}"); + Assert.True(node["ewma_ms"]!.GetValue() < 250, $"ewma {node["ewma_ms"]}"); } [Fact] @@ -1066,7 +1073,8 @@ public async Task A_burst_with_a_hanging_first_node_fails_over_in_parallel_not_i })); sw.Stop(); Assert.All(results, x => Assert.True(x.r.Outcome == SsrRpc.Outcome.Miss, $"{x.r.Outcome}: {x.r.Error}")); - Assert.All(results, x => Assert.True(x.ElapsedMilliseconds < 2 * 300, $"one call took {x.ElapsedMilliseconds} ms")); + // Queued behind each other the last call would land near 8 x 300 ms. + Assert.True(results.Max(x => x.ElapsedMilliseconds) < 1200, $"slowest call took {results.Max(x => x.ElapsedMilliseconds)} ms"); Assert.Equal(n, healthy.Methods.Count(m => m == "bridge.get_post_v2")); } @@ -1090,7 +1098,7 @@ async Task Burst(string author) })); // One round trip each: none of them waited for another's answer. Assert.All(results, x => Assert.True(x.r.Outcome == SsrRpc.Outcome.Miss, $"{x.r.Outcome}: {x.r.Error}")); - Assert.All(results, x => Assert.True(x.ElapsedMilliseconds < 400 + 200, $"one call took {x.ElapsedMilliseconds} ms")); + Assert.All(results, x => Assert.True(x.ElapsedMilliseconds < 750, $"one call took {x.ElapsedMilliseconds} ms")); } await Burst("cold"); diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index f5b9e28d..5e39ef9e 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs @@ -259,7 +259,7 @@ private IEnumerable UnsupportedFamilies(int nodeIndex) => var started = NowMs; try { - var useFallback = fallbackBody != null && Lacks(nodeIndex, family ?? method); + var useFallback = fallbackBody != null && Lacks(nodeIndex, family!); JsonNode? result; try { @@ -272,7 +272,7 @@ private IEnumerable UnsupportedFamilies(int nodeIndex) => // Two round trips against the caller's budget, but only once // per (node, family) per TTL for each call that reached the // node before the mark, and inside a fill the gate bounds. - _lacksUntil[(nodeIndex, family ?? method)] = _clock() + UnsupportedMethodTtlMs; + _lacksUntil[(nodeIndex, family!)] = _clock() + UnsupportedMethodTtlMs; // Rank the node on the v1 answer it will give from now on, // not on the probe that does not recur within the TTL. started = NowMs;