diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 7741ca6e..17e096dd 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -27,7 +27,18 @@ 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). + public HashSet Unknown = new(); + public readonly System.Collections.Concurrent.ConcurrentQueue Methods = new(); + public readonly System.Collections.Concurrent.ConcurrentQueue Params = new(); public RpcStub() { @@ -48,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)) @@ -72,17 +91,36 @@ 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 = 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 + "\"}}}}" + : 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}" : "{\"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 { } + // 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 { } } } @@ -624,6 +662,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", }) { Assert.True(SsrRpc.Allowlist.ContainsKey(key), key); @@ -631,6 +671,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] @@ -646,7 +690,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 +761,379 @@ 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_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")); + + 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(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[] { SsrRpc.V2Family }, unsupported); + } + finally + { + SsrRpc.SecretDigest = null; + } + } + + [Fact] + 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" }) + { + 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); + } + } + + [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(SsrRpc.V2Family, 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"]}"); + } + + [Fact] + 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, + // 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.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 < 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() < 250, $"ewma {node["ewma_ms"]}"); + } + + [Fact] + public async Task A_burst_with_a_hanging_first_node_fails_over_in_parallel_not_in_a_queue() + { + // 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}")); + // 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")); + } + + [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; + + 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 < 750, $"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() + { + 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()); + } + } diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index 506d57d9..495098e0 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,16 @@ 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). + // 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", "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 +330,22 @@ 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. + // The four v2 reads shipped together in one hivemind release, so a node + // lacking one lacks them all: one mark covers the family. + internal const string V2Family = "bridge.*_v2"; + + 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 +403,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), V2Family) : null); } finally { @@ -428,7 +455,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 40be5b1d..5e39ef9e 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,9 @@ 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 and the method's family +/// 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 +31,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 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. public HiveRpcClient(string[] nodes, int timeoutMs = 2000, int failoverThreshold = 2) @@ -43,6 +52,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 +92,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); + // Families this node is asked the fallback for until its mark lapses. + var unsupported = new JsonArray(); + foreach (var m in UnsupportedFamilies(v.Index)) unsupported.Add(m); + node["unsupported"] = unsupported; arr.Add(node); } return arr; @@ -98,6 +112,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 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; + // ---- calls ------------------------------------------------------------- /// Optional shape check for the RPC result. A node @@ -145,29 +166,82 @@ 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, 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 + /// . 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, Func? preferResult = null, - CallClass callClass = CallClass.Cheap) + CallClass callClass = CallClass.Cheap, + (string Method, JsonNode Params, string Family)? 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); + // 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() + { + ["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). + // 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 Lacks(int nodeIndex, string family) + { + 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; + } + + 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, Func? preferResult, - CallClass callClass) + CallClass callClass, + 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. var body = JsJson.Stringify(request); + var fallbackBody = fallbackRequest is null ? null : JsJson.Stringify(fallbackRequest); Exception? lastError = null; JsonNode? unpreferred = null; @@ -185,7 +259,25 @@ public RpcException(string message) : base(message) { } var started = NowMs; try { - var result = await CallNode(node, body); + var useFallback = fallbackBody != null && Lacks(nodeIndex, family!); + JsonNode? result; + try + { + result = await CallNode(node, useFallback ? fallbackBody! : body); + } + 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 for each call that reached the + // node before the mark, and inside a fill the gate bounds. + _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; + result = await CallNode(node, fallbackBody); + } if (validateResult != null && !validateResult(result)) { // Don't burn a same-node retry: a node serving malformed