diff --git a/dotnet/EcencyApi.Tests/HiveRpcFailoverTests.cs b/dotnet/EcencyApi.Tests/HiveRpcFailoverTests.cs index cc239e44..6a381e9d 100644 --- a/dotnet/EcencyApi.Tests/HiveRpcFailoverTests.cs +++ b/dotnet/EcencyApi.Tests/HiveRpcFailoverTests.cs @@ -179,6 +179,225 @@ public async Task TimeoutNode_FailsOverToHealthyNode() Assert.Equal("served-by", result![0]!["name"]!.GetValue()); } + [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); + var goodStatus = 200; + await using var good = new StubNode(() => goodStatus); + + 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()); + Assert.Equal(3, view["timeouts"]!.GetValue()); + Assert.Equal(3, view["samples"]!.GetValue()); // the timeouts ARE latency samples + // ...floored at the unproven prior: a 300ms timeout must not make a + // node that never answered look faster than nodes never tried. + Assert.True(view["ewma_ms"]!.GetValue() > 1000); + Assert.True(view["parked_for_ms"]!.GetValue() > 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()); + Assert.True(good.Hits >= 1); + + // The park lapses (30s). Its timeouts were recorded as latency (floored + // above the unproven prior), so the dead node now ranks behind the + // proven leader AND behind the never-tried node: a leader hiccup goes + // to the untried node, not to it. + now += 31_000; + flakyStatus = 500; + await client.Call("condenser_api", "get_accounts", new JsonArray()); + Assert.Equal(3, client.HealthSnapshot()[0]!["calls"]!.GetValue()); + + // Only when every other node fails is it probed: one attempt, which + // fails and re-parks it for twice as long. + flakyStatus = 500; + goodStatus = 500; + await Assert.ThrowsAnyAsync(() => client.Call("condenser_api", "get_accounts", new JsonArray())); + flakyStatus = 200; + goodStatus = 200; + view = client.HealthSnapshot()[0]!; + Assert.Equal(4, view["calls"]!.GetValue()); + Assert.True(view["parked_for_ms"]!.GetValue() > 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()); + } + + [Fact] + public async Task ParkingEndsTheSameNodeRetry_WithTheDefaultFailoverThreshold() + { + // failoverThreshold 2 (the default client) retries a node once before + // moving on. The failure that parks the node must end that retry too, + // or the parked node gets one more attempt, one more timeout, and a + // park that starts out twice as long. + await using var dead = new StubNode(() => -1); + await using var good = new StubNode(() => 200); + long now = 0; + var client = new HiveRpcClient(new[] { dead.Url, good.Url }, timeoutMs: 200, failoverThreshold: 2, clock: () => now); + + await client.Call("condenser_api", "get_accounts", new JsonArray()); // dead: 2 attempts, then good + var view = client.HealthSnapshot()[0]!; + Assert.Equal(2, view["calls"]!.GetValue()); + Assert.Equal(0, view["parked_for_ms"]!.GetValue()); + + now += 31_000; + await client.Call("condenser_api", "get_accounts", new JsonArray()); // third failure parks: ONE attempt + view = client.HealthSnapshot()[0]!; + Assert.Equal(3, view["calls"]!.GetValue()); + Assert.Equal(30_000, view["parked_for_ms"]!.GetValue()); + } + + [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(() => client.Call("condenser_api", "get_accounts", new JsonArray())); + } + Assert.Equal(3, client.HealthSnapshot()[0]!["calls"]!.GetValue()); + Assert.True(client.HealthSnapshot()[0]!["parked_for_ms"]!.GetValue() > 0); + await Assert.ThrowsAnyAsync(() => client.Call("condenser_api", "get_accounts", new JsonArray())); + Assert.Equal(4, client.HealthSnapshot()[0]!["calls"]!.GetValue()); + } + + [Fact] + public async Task ANodeOutOfItsPark_AdmitsOneProbeAtATime() + { + // Calls in flight when a park lapses all still hold the node in the + // ordering they took at their start. It ranks last, so they reach it + // only when every other node has failed them; when that happens to a + // whole burst at once, only ONE call may probe the recovering node. The + // rest fail fast instead of each paying the timeout against a node that + // was not answering a moment ago. This is the burst the parking exists + // to prevent, seen from the other side of the park. + await using var dead = new StubNode(() => -1); + var flakyStatus = 200; + await using var flaky = new StubNode(() => flakyStatus); + var goodStatus = 200; + await using var good = new StubNode(() => goodStatus); + long now = 0; + var client = new HiveRpcClient(new[] { dead.Url, flaky.Url, good.Url }, timeoutMs: 300, failoverThreshold: 1, clock: () => now); + + for (var i = 0; i < 3; i++) + { + if (i > 0) now += 31_000; + await client.Call("condenser_api", "get_accounts", new JsonArray()); + } + Assert.True(client.HealthSnapshot()[0]!["parked_for_ms"]!.GetValue() > 0); + now += 31_000; // the park lapses: half-open + + // The leaders throttle (429): responsive, failing every call, and never + // failure-parked by it, so the pool is not "entirely down" and the + // half-open rule is what decides who reaches the recovering node. + flakyStatus = 429; + goodStatus = 429; + var burst = Enumerable.Range(0, 12) + .Select(async _ => + { + try { await client.Call("condenser_api", "get_accounts", new JsonArray()); return true; } + catch { return false; } + }) + .ToArray(); + var served = await Task.WhenAll(burst); + flakyStatus = 200; + goodStatus = 200; + Assert.All(served, ok => Assert.False(ok)); // nothing answered: the dead node is dead + var view = client.HealthSnapshot()[0]!; + Assert.Equal(4, view["calls"]!.GetValue()); // exactly one probe out of twelve + Assert.True(view["parked_for_ms"]!.GetValue() > 30_000); // which re-parked it for longer + + // With the leaders back, the re-parked node is skipped. + await client.Call("condenser_api", "get_accounts", new JsonArray()); + Assert.Equal(4, client.HealthSnapshot()[0]!["calls"]!.GetValue()); + } + + [Fact] + public async Task RateLimits_DoNotCountTowardFailureParking() + { + // 429s have their own parking and a throttled node is responsive: two + // 429s and one hard failure must not hard-park it. + var status = 429; + await using var throttled = new StubNode(() => status); + await using var good = new StubNode(() => 200); + long now = 0; + var client = new HiveRpcClient(new[] { throttled.Url, good.Url }, timeoutMs: 500, failoverThreshold: 1, clock: () => now); + + await client.Call("condenser_api", "get_accounts", new JsonArray()); // 429 -> rate-limit parked + now += 61_000; // that park lapses + await client.Call("condenser_api", "get_accounts", new JsonArray()); // 429 again + now += 61_000; + status = 500; + await client.Call("condenser_api", "get_accounts", new JsonArray()); // one hard failure + var view = client.HealthSnapshot()[0]!; + Assert.Equal(3, view["calls"]!.GetValue()); + Assert.Equal(2, view["rate_limited"]!.GetValue()); + Assert.Equal(0, view["parked_for_ms"]!.GetValue()); + } + + [Fact] + public async Task ASuccessClearsAFailurePark() + { + // An all-parked pool offers every node; the one that recovers must not + // be excluded again by its stale park deadline once another node's + // park lapses. + var mode = -1; + await using var flapping = new StubNode(() => mode); + long now = 0; + var client = new HiveRpcClient(new[] { flapping.Url }, timeoutMs: 200, failoverThreshold: 1, clock: () => now); + for (var i = 0; i < 3; i++) + { + await Assert.ThrowsAnyAsync(() => client.Call("condenser_api", "get_accounts", new JsonArray())); + } + Assert.True(client.HealthSnapshot()[0]!["parked_for_ms"]!.GetValue() > 0); + // The only node, so it is still offered; it recovers and answers. + mode = 200; + await Task.Delay(3500); // let the stub's hung handlers drain before it can answer + await client.Call("condenser_api", "get_accounts", new JsonArray()); + Assert.Equal(0, client.HealthSnapshot()[0]!["parked_for_ms"]!.GetValue()); + Assert.Equal(0, client.HealthSnapshot()[0]!["consecutive_failures"]!.GetValue()); + } + + [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() { diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 027d4194..239f1371 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -436,6 +436,73 @@ public async Task With_the_secret_configured_both_routes_serve_the_matching_head } } + [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()); + Assert.Equal(1, post["slow_fill"]!.GetValue()); + Assert.Equal(4, post["coalesced"]!.GetValue()); + Assert.Equal(1, stub.Hits); + + var nodes = body["nodes"]!.AsArray(); + var node = Assert.Single(nodes)!; + Assert.Equal("127.0.0.1", node["node"]!.GetValue()); + Assert.Equal(1, node["calls"]!.GetValue()); + Assert.Equal(1, node["ok"]!.GetValue()); + Assert.Equal(0, node["timeouts"]!.GetValue()); + Assert.Equal(0, node["parked_for_ms"]!.GetValue()); + } + finally + { + SsrRpc.SecretDigest = null; + SsrRpc.BudgetMs = 1500; + } + } + + [Fact] + public async Task A_slow_fill_that_then_fails_is_still_a_slow_fill() + { + // During an upstream outage every reader times out and the fill ends + // in an error; that is precisely when the slow-fill count must not + // read zero. + await using var stub = new RpcStub { DelayMs = 400, RpcError = true }; + Use(stub, budgetMs: 100); + SsrRpc.SecretDigest = SsrRpc.Digest("right-secret"); + try + { + var r = await SsrRpc.Resolve(Post, P("slow", "broken")); + Assert.Equal(SsrRpc.Outcome.Timeout, r.Outcome); + await Task.Delay(600); + var stats = Request("GET", "/private-api/ssr/stats", "right-secret"); + await SsrRpc.Stats(stats); + var post = JsonNode.Parse(ResponseText(stats))!["methods"]!["bridge.get_post"]!; + Assert.Equal(1, post["slow_fill"]!.GetValue()); + Assert.Equal(1, post["timeout"]!.GetValue()); + } + finally + { + SsrRpc.SecretDigest = null; + SsrRpc.BudgetMs = 1500; + } + } + [Fact] public async Task A_null_result_is_served_as_json_null_with_a_json_content_type() { diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index 811c1d03..5b2c91aa 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/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(); @@ -349,12 +353,22 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key } } var started = Environment.TickCount64; - // The dotted method form: hived's legacy `call` dispatcher has no API - // named `bridge` (found on alpha: every bridge read failed with - // "Could not find API bridge"), while `bridge.get_post` is routed to - // hivemind. The node already hangs off the request body and cannot be - // re-parented into the envelope, so it travels as a clone. - var result = await Client.CallMethod($"{policy.Api}.{policy.Method}", @params.DeepClone()); + JsonNode? result; + try + { + // The dotted method form: hived's legacy `call` dispatcher has no API + // named `bridge` (found on alpha: every bridge read failed with + // "Could not find API bridge"), while `bridge.get_post` is routed to + // 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()); + } + finally + { + // A fill that outran the budget is slow whether it then landed or + // threw: an upstream outage is exactly when the count matters. + if (Environment.TickCount64 - started > BudgetMs) Interlocked.Increment(ref counter.SlowFill); + } var bytes = Encoding.UTF8.GetBytes(result is null ? "null" : JsJson.Stringify(result)); counter.RecordUpstream(Environment.TickCount64 - started); Cache.Set(key, bytes, policy.TtlMs); @@ -444,6 +458,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 +473,7 @@ public static async Task Stats(HttpContext ctx) }, ["budget_ms"] = BudgetMs, ["methods"] = methods, + ["nodes"] = Client.HealthSnapshot(), }); } } diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index e58447fe..8678bb70 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/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) + { + } + + /// Test seam for the health tracker's notion of time. + internal HiveRpcClient(string[] nodes, int timeoutMs, int failoverThreshold, Func? clock) { _nodes = nodes; _timeoutMs = timeoutMs; _failoverThreshold = Math.Max(1, failoverThreshold); - _health = new NodeHealthTracker(nodes.Length); + _health = new NodeHealthTracker(nodes.Length, clock); + } + + public IReadOnlyList Nodes => _nodes; + + /// + /// 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. + /// + 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 @@ -121,6 +158,8 @@ public RpcException(string message) : base(message) { } for (var attempt = 0; attempt < _failoverThreshold; attempt++) { + // The ordering above is a snapshot; admission is decided now. + if (!_health.TryBeginAttempt(nodeIndex)) break; var started = NowMs; try { @@ -171,9 +210,9 @@ public RpcException(string message) : base(message) { } { _health.RecordRateLimited(nodeIndex, e.RetryAfterMs); } - else + else if (_health.RecordFailure(nodeIndex, NowMs - started, e.IsTimeout)) { - _health.RecordFailure(nodeIndex, NowMs - started); + break; // this failure parked the node: no same-node retry } if (e.AdvanceImmediately) { @@ -183,7 +222,14 @@ public RpcException(string message) : base(message) { } catch (Exception e) { lastError = e; - _health.RecordFailure(nodeIndex, NowMs - started); + if (_health.RecordFailure(nodeIndex, NowMs - started)) + { + break; + } + } + finally + { + _health.EndAttempt(nodeIndex); } } } @@ -204,15 +250,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 +285,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 +412,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 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", diff --git a/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs b/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs index f4450e46..c4395ad0 100644 --- a/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs +++ b/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs @@ -30,6 +30,16 @@ public sealed class NodeHealthTracker private const int LatencyMaxAgeMs = 5 * 60_000; private const double LatencyUnprovenPriorMs = 1_000; private const int SlowFailureFloorMs = 2_000; + // A node that fails this many times in a row is parked (30s, doubling to + // 120s; a success clears the streak) and is not tried while any other node + // is available. Without this a node that never answers keeps its + // "unexplored" standing: only 429s parked, and a failure demoted for 30s at + // most, so each time the leading nodes hiccupped the dead node took the + // whole in-flight burst at the full per-node timeout (observed as dozens + // of half-open connects at once to one unreachable node). + private const int FailureParkThreshold = 3; + private const int FailureParkBaseMs = 30_000; + private const int FailureParkMaxMs = 120_000; private sealed class NodeHealth { @@ -38,24 +48,42 @@ private sealed class NodeHealth public long RateLimitedUntilMs; public int RateLimitStreak; public long LastRateLimitAtMs; + public long FailureParkedUntilMs; + public int FailureParkStreak; + // Hard failures only (timeouts, refusals, bad answers), never 429s: a + // throttled node is responsive and has its own parking. + public int ConsecutiveHardFailures; + // Attempts currently in flight against this node (a gauge, for the + // half-open rule below). + public int InFlight; public double? EwmaLatencyMs; public int LatencySampleCount; public long LatencyUpdatedAtMs; + // Lifetime counters, for the stats endpoint. + public long Calls, Successes, Failures, Timeouts, RateLimits; } + /// One node's health as reported by . + public sealed record NodeView( + int Index, long Calls, long Successes, long Failures, long Timeouts, long RateLimits, + double? EwmaLatencyMs, int LatencySamples, int ConsecutiveFailures, + bool RecentFailure, long RateLimitedForMs, long FailureParkedForMs); + private readonly NodeHealth[] _health; private readonly object _lock = new(); + private readonly Func _clock; - public NodeHealthTracker(int nodeCount) + public NodeHealthTracker(int nodeCount, Func? clock = null) { _health = new NodeHealth[nodeCount]; for (var i = 0; i < nodeCount; i++) { _health[i] = new NodeHealth(); } + _clock = clock ?? (() => Environment.TickCount64); } - private static long NowMs => Environment.TickCount64; + private long NowMs => _clock(); // ---- health bookkeeping (lock-guarded; contention is negligible) ------ @@ -64,25 +92,57 @@ public void RecordSuccess(int nodeIndex, double elapsedMs) lock (_lock) { var h = _health[nodeIndex]; + h.Calls++; + h.Successes++; h.ConsecutiveFailures = 0; + h.ConsecutiveHardFailures = 0; h.RateLimitStreak = 0; + h.FailureParkStreak = 0; + // A node that just answered is not parked, whatever the deadline said: + // an all-parked pool offers every node, and the one that recovers must + // not be pushed out again by a stale deadline the moment another + // node's park lapses. + h.FailureParkedUntilMs = 0; RecordLatency(h, elapsedMs); } } - public void RecordFailure(int nodeIndex, double elapsedMs) + /// The attempt ran into the client's per-node timeout. + /// That IS a latency sample whatever the timeout is set to; the floor below + /// only tells instant refusals (a down node is not "slow") from slow 5xx. + /// True when this failure parked the node (or it was already + /// parked): the caller should not retry it, a same-node retry would only + /// add another timeout and lengthen the park. + public bool RecordFailure(int nodeIndex, double elapsedMs, bool timedOut = false) { lock (_lock) { var h = _health[nodeIndex]; + var now = NowMs; + h.Calls++; + h.Failures++; + if (timedOut) h.Timeouts++; h.ConsecutiveFailures++; - h.LastFailureAtMs = NowMs; - // A genuinely slow failure (timeout / slow 5xx) is also a latency - // signal; an instant refusal is not (a *down* node isn't "slow"). - if (elapsedMs >= SlowFailureFloorMs) + h.ConsecutiveHardFailures++; + h.LastFailureAtMs = now; + if (timedOut) + { + // A timeout says "at least this slow". Floored at the unproven + // prior so a short client timeout cannot rank a node that never + // answered ahead of nodes that were never tried. + RecordLatency(h, Math.Max(elapsedMs, LatencyUnprovenPriorMs + 1)); + } + else if (elapsedMs >= SlowFailureFloorMs) { RecordLatency(h, elapsedMs); } + if (h.ConsecutiveHardFailures >= FailureParkThreshold) + { + var parkMs = Math.Min(FailureParkBaseMs << Math.Min(h.FailureParkStreak, 2), FailureParkMaxMs); + h.FailureParkedUntilMs = now + parkMs; + h.FailureParkStreak++; + } + return h.FailureParkedUntilMs > now; } } @@ -92,6 +152,8 @@ public void RecordRateLimited(int nodeIndex, int? retryAfterMs) { var h = _health[nodeIndex]; var now = NowMs; + h.Calls++; + h.RateLimits++; h.ConsecutiveFailures++; h.LastFailureAtMs = now; if (now - h.LastRateLimitAtMs > RateLimitStreakResetMs) @@ -106,7 +168,7 @@ public void RecordRateLimited(int nodeIndex, int? retryAfterMs) } } - private static void RecordLatency(NodeHealth h, double elapsedMs) + private void RecordLatency(NodeHealth h, double elapsedMs) { var now = NowMs; // A stale profile restarts from scratch so an idle process re-learns @@ -125,28 +187,38 @@ private static void RecordLatency(NodeHealth h, double elapsedMs) /// /// Node indices ordered best-first: unparked nodes sorted by - /// (recent-failure tier, latency score, config index); parked - /// (rate-limited) nodes appended last as a final resort. A recovered node - /// re-enters the healthy tiers as soon as its windows lapse. + /// (recent-failure tier, latency score, config index); rate-limit-parked + /// nodes appended last as a final resort. A node parked for consecutive + /// failures is left out altogether while any other node is available (it + /// was not answering; a throttled node might), and probed once its park + /// lapses. When every node is failure-parked all are offered, so a pool + /// that is entirely down degrades to "try them" rather than "try nothing". /// public List OrderedNodeIndices() { lock (_lock) { var now = NowMs; - return Enumerable.Range(0, _health.Length) + var ranked = Enumerable.Range(0, _health.Length) .Select(i => { var h = _health[i]; var parked = h.RateLimitedUntilMs > now; + var dead = h.FailureParkedUntilMs > now; var recentFailure = h.ConsecutiveFailures > 0 && now - h.LastFailureAtMs < RecentFailureWindowMs; var latencyUsable = h.EwmaLatencyMs is not null && h.LatencySampleCount >= LatencyMinSamples && now - h.LatencyUpdatedAtMs <= LatencyMaxAgeMs; var score = latencyUsable ? h.EwmaLatencyMs!.Value : LatencyUnprovenPriorMs; - return (Index: i, Parked: parked, RecentFailure: recentFailure, Score: score); + return (Index: i, Parked: parked, Dead: dead, RecentFailure: recentFailure, Score: score); }) + .ToList(); + if (ranked.Any(x => !x.Dead)) + { + ranked.RemoveAll(x => x.Dead); + } + return ranked .OrderBy(x => x.Parked) .ThenBy(x => x.RecentFailure) .ThenBy(x => x.Score) @@ -156,6 +228,67 @@ public List OrderedNodeIndices() } } + /// + /// Admission at attempt time, because the ordering a call holds was taken + /// when the call started and can be stale by the time it reaches this node: + /// a node parked since then is skipped, and a node just out of a park is + /// half-open, admitting one probe at a time, so a burst of concurrent calls + /// that all hold it in their lists cannot all probe it at once. Either + /// rule yields when no other node could take the attempt. Pair with + /// . + /// + public bool TryBeginAttempt(int nodeIndex) + { + lock (_lock) + { + var now = NowMs; + var h = _health[nodeIndex]; + var parked = h.FailureParkedUntilMs > now; + var halfOpenBusy = !parked && h.FailureParkStreak > 0 && h.InFlight > 0; + if (parked || halfOpenBusy) + { + var othersAvailable = false; + for (var j = 0; j < _health.Length && !othersAvailable; j++) + { + if (j == nodeIndex) continue; + var o = _health[j]; + var oParked = o.FailureParkedUntilMs > now; + var oBusy = !oParked && o.FailureParkStreak > 0 && o.InFlight > 0; + othersAvailable = !oParked && !oBusy; + } + if (othersAvailable) return false; + } + h.InFlight++; + return true; + } + } + + public void EndAttempt(int nodeIndex) + { + lock (_lock) + { + var h = _health[nodeIndex]; + if (h.InFlight > 0) h.InFlight--; + } + } + + /// Per-node state and lifetime counters, for the stats endpoint. + public List Snapshot() + { + lock (_lock) + { + var now = NowMs; + return Enumerable.Range(0, _health.Length).Select(i => + { + var h = _health[i]; + return new NodeView(i, h.Calls, h.Successes, h.Failures, h.Timeouts, h.RateLimits, + h.EwmaLatencyMs, h.LatencySampleCount, h.ConsecutiveFailures, + h.ConsecutiveFailures > 0 && now - h.LastFailureAtMs < RecentFailureWindowMs, + Math.Max(0, h.RateLimitedUntilMs - now), Math.Max(0, h.FailureParkedUntilMs - now)); + }).ToList(); + } + } + /// Retry-After: delta-seconds or an HTTP date (RFC 9110). public static int? ParseRetryAfterMs(string? header) {