RPC pool: park a node that keeps failing, drop the unreachable one, per-node stats

dotnet/EcencyApi.Tests/HiveRpcFailoverTests.cs

@@ -179,6 +179,93 @@ public async Task TimeoutNode_FailsOverToHealthyNode()
         Assert.Equal("served-by", result![0]!["name"]!.GetValue<string>());
     }
 
+    [Fact]
+    public async Task DeadNode_IsParkedAfterThreeTimeouts_AndSkippedWhileOthersServe()
+    {
+        // A node that never answers used to keep its "unexplored" standing
+        // (timeouts below the slow-failure floor left no latency sample, and a
+        // failure only demoted it for 30s), so it was retried every window and
+        // took whole bursts of concurrent calls. Now: a timeout is a latency
+        // sample, three failures in a row park it, and a parked node is skipped
+        // while any other node is available.
+        await using var dead = new StubNode(() => -1);      // hangs past the timeout
+        var flakyStatus = 200;
+        await using var flaky = new StubNode(() => flakyStatus);
+        await using var good = new StubNode(() => 200);
+
+        long now = 0;
+        var client = new HiveRpcClient(new[] { dead.Url, flaky.Url, good.Url }, timeoutMs: 300, failoverThreshold: 1, clock: () => now);
+
+        // Three calls, each past the previous one's 30s recent-failure window:
+        // the dead node (config order, all unproven) is tried first every time.
+        for (var i = 0; i < 3; i++)
+        {
+            if (i > 0) now += 31_000;
+            await client.Call("condenser_api", "get_accounts", new JsonArray());
+        }
+        // The stub's accept loop is single-threaded and its hang outlives the
+        // client timeout, so later attempts queue in the listener backlog and
+        // never reach its hit counter; the client's own per-node counters are
+        // the measure of what was attempted.
+        var view = client.HealthSnapshot()[0]!;
+        Assert.Equal(3, view["calls"]!.GetValue<long>());
+        Assert.Equal(3, view["timeouts"]!.GetValue<long>());
+        Assert.Equal(3, view["samples"]!.GetValue<int>());       // the timeouts ARE latency samples
+        Assert.True(view["ewma_ms"]!.GetValue<double>() >= 250);
+        Assert.True(view["parked_for_ms"]!.GetValue<long>() > 0); // parked after the third
+
+        // The leader hiccups: with the dead node parked, the call skips it and
+        // fails over from flaky straight to good instead of handing the dead
+        // node the burst.
+        flakyStatus = 500;
+        await client.Call("condenser_api", "get_accounts", new JsonArray());
+        flakyStatus = 200;
+        Assert.Equal(3, client.HealthSnapshot()[0]!["calls"]!.GetValue<long>());
+        Assert.True(good.Hits >= 1);
+
+        // The park lapses (30s). Its timeouts were recorded as latency, so the
+        // dead node now ranks behind the proven-fast leader and is only probed
+        // when the leader fails: one probe, which fails and re-parks it for
+        // twice as long; the call is still served by the third node.
+        now += 31_000;
+        flakyStatus = 500;
+        await client.Call("condenser_api", "get_accounts", new JsonArray());
+        flakyStatus = 200;
+        view = client.HealthSnapshot()[0]!;
+        Assert.Equal(4, view["calls"]!.GetValue<long>());
+        Assert.True(view["parked_for_ms"]!.GetValue<long>() > 30_000);
+
+        // Inside the doubled park no probe is made, even past the recent-failure window.
+        now += 45_000;
+        await client.Call("condenser_api", "get_accounts", new JsonArray());
+        Assert.Equal(4, client.HealthSnapshot()[0]!["calls"]!.GetValue<long>());
+    }
+
+    [Fact]
+    public async Task AllNodesFailureParked_AreStillTried()
+    {
+        // A pool that is entirely parked degrades to "try them", never to
+        // "try nothing": the caller gets the node error, not a synthetic one.
+        await using var dead = new StubNode(() => -1);
+        long now = 0;
+        var client = new HiveRpcClient(new[] { dead.Url }, timeoutMs: 200, failoverThreshold: 1, clock: () => now);
+        for (var i = 0; i < 3; i++)
+        {
+            await Assert.ThrowsAnyAsync<Exception>(() => client.Call("condenser_api", "get_accounts", new JsonArray()));
+        }
+        Assert.Equal(3, client.HealthSnapshot()[0]!["calls"]!.GetValue<long>());
+        Assert.True(client.HealthSnapshot()[0]!["parked_for_ms"]!.GetValue<long>() > 0);
+        await Assert.ThrowsAnyAsync<Exception>(() => client.Call("condenser_api", "get_accounts", new JsonArray()));
+        Assert.Equal(4, client.HealthSnapshot()[0]!["calls"]!.GetValue<long>());
+    }
+
+    [Fact]
+    public void DefaultPool_DoesNotCarryTheUnreachableNode()
+    {
+        Assert.DoesNotContain(HiveClients.DefaultNodes, n => n.Contains("arcange", StringComparison.Ordinal));
+        Assert.True(HiveClients.DefaultNodes.Count >= 6);
+    }
+
     [Fact]
     public async Task ProvenSlowNode_IsDemotedByLatencyEwma()
     {

dotnet/EcencyApi.Tests/SsrRpcTests.cs

@@ -120,7 +120,7 @@
         var tasks = Enumerable.Range(0, 12).Select(_ => SsrRpc.Resolve(Post, P("a", "b"))).ToArray();
         var results = await Task.WhenAll(tasks);
         Assert.Equal(1, stub.Hits);
         Assert.Single(results.Where(r => r.Outcome == SsrRpc.Outcome.Miss));
         Assert.Equal(11, results.Count(r => r.Outcome == SsrRpc.Outcome.Coalesced));
         var bodies = results.Select(r => Encoding.UTF8.GetString(r.Bytes)).Distinct().ToArray();
         Assert.Single(bodies);
@@ -349,8 +349,8 @@
         await Task.Delay(180);
         var c = SsrRpc.Resolve(Post, P("k", "2"));   // fresh reader at ~190ms, coalesces
         await Task.WhenAll(a, b, c);
         Assert.Equal(SsrRpc.Outcome.Timeout, a.Result.Outcome);
         Assert.Equal(SsrRpc.Outcome.Timeout, b.Result.Outcome);
         // Judged by the fresh attach, not by the original enqueue: the fill ran.
         await Task.Delay(500);
         Assert.Equal(2, stub.Hits);
@@ -436,6 +436,46 @@
         }
     }
 
+    [Fact]
+    public async Task Stats_count_one_slow_fill_for_many_waiter_timeouts_and_report_per_node_health()
+    {
+        // `timeout` is per waiting reader; `slow_fill` is per fill. A hot key
+        // with five readers on one slow upstream call is five timeouts and one
+        // slow fill, and the per-node section shows which node served it.
+        await using var stub = new RpcStub { DelayMs = 400 };
+        Use(stub, budgetMs: 100);
+        SsrRpc.SecretDigest = SsrRpc.Digest("right-secret");
+        try
+        {
+            var readers = Enumerable.Range(0, 5).Select(_ => SsrRpc.Resolve(Post, P("hot", "key"))).ToArray();
+            var outcomes = await Task.WhenAll(readers);
+            Assert.All(outcomes, r => Assert.Equal(SsrRpc.Outcome.Timeout, r.Outcome));
+            await Task.Delay(600); // the detached fill completes and lands
+
+            var stats = Request("GET", "/private-api/ssr/stats", "right-secret");
+            await SsrRpc.Stats(stats);
+            var body = JsonNode.Parse(ResponseText(stats))!;
+            var post = body["methods"]!["bridge.get_post"]!;
+            Assert.Equal(5, post["timeout"]!.GetValue<long>());
+            Assert.Equal(1, post["slow_fill"]!.GetValue<long>());
+            Assert.Equal(4, post["coalesced"]!.GetValue<long>());
+            Assert.Equal(1, stub.Hits);
+
+            var nodes = body["nodes"]!.AsArray();
+            var node = Assert.Single(nodes)!;
+            Assert.Equal("127.0.0.1", node["node"]!.GetValue<string>());
+            Assert.Equal(1, node["calls"]!.GetValue<long>());
+            Assert.Equal(1, node["ok"]!.GetValue<long>());
+            Assert.Equal(0, node["timeouts"]!.GetValue<long>());
+            Assert.Equal(0, node["parked_for_ms"]!.GetValue<long>());
+        }
+        finally
+        {
+            SsrRpc.SecretDigest = null;
+            SsrRpc.BudgetMs = 1500;
+        }
+    }
+
     [Fact]
     public async Task A_null_result_is_served_as_json_null_with_a_json_content_type()
     {

dotnet/EcencyApi/Handlers/SsrRpc.cs

@@ -128,6 +128,10 @@ public bool TryExpire(int budgetMs)
     internal sealed class Counter
     {
         public long Hit, Miss, Coalesced, Error, Timeout;
+        // Fills that outran the lookup budget. Distinct from Timeout, which is
+        // counted once per waiting reader: one slow fill on a hot key is one
+        // slow fill and many timeouts.
+        public long SlowFill;
         // Upstream latency EWMA for misses, milliseconds.
         public double UpstreamMs;
         private readonly object _lock = new();
@@ -356,7 +360,9 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key
             // re-parented into the envelope, so it travels as a clone.
             var result = await Client.CallMethod($"{policy.Api}.{policy.Method}", @params.DeepClone());
             var bytes = Encoding.UTF8.GetBytes(result is null ? "null" : JsJson.Stringify(result));
-            counter.RecordUpstream(Environment.TickCount64 - started);
+            var elapsed = Environment.TickCount64 - started;
+            counter.RecordUpstream(elapsed);
+            if (elapsed > BudgetMs) Interlocked.Increment(ref counter.SlowFill);
             Cache.Set(key, bytes, policy.TtlMs);
             tcs.TrySetResult(bytes);
         }
@@ -444,6 +450,7 @@ public static async Task Stats(HttpContext ctx)
                 ["coalesced"] = Interlocked.Read(ref c.Coalesced),
                 ["error"] = Interlocked.Read(ref c.Error),
                 ["timeout"] = Interlocked.Read(ref c.Timeout),
+                ["slow_fill"] = Interlocked.Read(ref c.SlowFill),
                 ["upstream_ms"] = Math.Round(c.ReadUpstreamMs(), 1),
             };
         }
@@ -458,6 +465,7 @@ public static async Task Stats(HttpContext ctx)
             },
             ["budget_ms"] = BudgetMs,
             ["methods"] = methods,
+            ["nodes"] = Client.HealthSnapshot(),
         });
     }
 }

dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs

@@ -28,11 +28,48 @@ public sealed class HiveRpcClient
     // 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)
+        : this(nodes, timeoutMs, failoverThreshold, null)
+    {
+    }
+
+    /// <param name="clock">Test seam for the health tracker's notion of time.</param>
+    internal HiveRpcClient(string[] nodes, int timeoutMs, int failoverThreshold, Func<long>? clock)
     {
         _nodes = nodes;
         _timeoutMs = timeoutMs;
         _failoverThreshold = Math.Max(1, failoverThreshold);
-        _health = new NodeHealthTracker(nodes.Length);
+        _health = new NodeHealthTracker(nodes.Length, clock);
+    }
+
+    public IReadOnlyList<string> Nodes => _nodes;
+
+    /// <summary>
+    /// Per-node health for an internal stats endpoint: which node carries the
+    /// traffic, which one times out, which one is parked. Hosts only; these are
+    /// the public node names, nothing about this deployment.
+    /// </summary>
+    public JsonArray HealthSnapshot()
+    {
+        var arr = new JsonArray();
+        foreach (var v in _health.Snapshot())
+        {
+            arr.Add(new JsonObject
+            {
+                ["node"] = Uri.TryCreate(_nodes[v.Index], UriKind.Absolute, out var u) ? u.Host : _nodes[v.Index],
+                ["calls"] = v.Calls,
+                ["ok"] = v.Successes,
+                ["failures"] = v.Failures,
+                ["timeouts"] = v.Timeouts,
+                ["rate_limited"] = v.RateLimits,
+                ["ewma_ms"] = v.EwmaLatencyMs is { } e ? Math.Round(e, 1) : null,
+                ["samples"] = v.LatencySamples,
+                ["consecutive_failures"] = v.ConsecutiveFailures,
+                ["recent_failure"] = v.RecentFailure,
+                ["rate_limited_for_ms"] = v.RateLimitedForMs,
+                ["parked_for_ms"] = v.FailureParkedForMs,
+            });
+        }
+        return arr;
     }
 
     public sealed class RpcException : Exception
@@ -173,7 +210,7 @@ public RpcException(string message) : base(message) { }
                     }
                     else
                     {
-                        _health.RecordFailure(nodeIndex, NowMs - started);
+                        _health.RecordFailure(nodeIndex, NowMs - started, e.IsTimeout);
                     }
                     if (e.AdvanceImmediately)
                     {
@@ -204,15 +241,17 @@ private sealed class NodeUnavailableException : Exception
     {
         public bool AdvanceImmediately { get; }
         public bool IsRateLimit { get; }
+        public bool IsTimeout { get; }
         public int? RetryAfterMs { get; }
         public Exception? Cause { get; private set; }
 
         public NodeUnavailableException(string message, bool advanceImmediately,
-            bool isRateLimit = false, int? retryAfterMs = null) : base(message)
+            bool isRateLimit = false, int? retryAfterMs = null, bool isTimeout = false) : base(message)
         {
             AdvanceImmediately = advanceImmediately;
             IsRateLimit = isRateLimit;
             RetryAfterMs = retryAfterMs;
+            IsTimeout = isTimeout;
         }
 
         public NodeUnavailableException WithInner(Exception inner) { Cause = inner; return this; }
@@ -237,7 +276,7 @@ public NodeUnavailableException(string message, bool advanceImmediately,
         }
         catch (OperationCanceledException e) when (cts.IsCancellationRequested)
         {
-            throw new NodeUnavailableException($"RPC node {node} timed out", advanceImmediately: false).WithInner(e);
+            throw new NodeUnavailableException($"RPC node {node} timed out", advanceImmediately: false, isTimeout: true).WithInner(e);
         }
         catch (HttpRequestException e)
         {
@@ -364,12 +403,14 @@ public static class HiveClients
     // portfolio engine/chain layers came back empty for everyone. GetAccounts
     // also routes around such a node at runtime, but keeping them out of the pool
     // means correctness here does not depend on that fallback firing.
+    // hive-api.arcange.eu is absent too: it never completes a TCP connect from
+    // any host this service runs on (SYN, no answer), so every attempt cost the
+    // full per-node timeout and, in bursts, took every in-flight fill with it.
     public static readonly IReadOnlyList<string> DefaultNodes = new[]
     {
         "https://api.hive.blog",
         "https://api.deathwing.me",
         "https://rpc.mahdiyari.info",
-        "https://hive-api.arcange.eu",
         "https://api.openhive.network",
         "https://hive-api.3speak.tv",
         "https://api.syncad.com",