diff --git a/CLAUDE.md b/CLAUDE.md index c5972da8..8142ce3e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -41,7 +41,9 @@ dotnet/EcencyApi/ Infrastructure/ Upstream.cs upstream HTTP + Express-compatible response piping ApiClient.cs private-API requests (PRIVATE_API_AUTH header injection) + NodeHealthTracker.cs per-node health state + pool ordering (shared by the RPC clients) HiveRpcClient.cs Hive RPC with health-aware node failover + EngineRpcClient.cs Hive-Engine /contracts with the same failover HiveCrypto.cs secp256k1: key-from-login, canonical signing, pubkey recovery JsJson.cs / JsVal.cs JavaScript-semantics layer (see invariants) B64u.cs, MemCache.cs, HttpContextExtensions.cs @@ -64,9 +66,14 @@ Handlers are `public static async Task Name(HttpContext ctx)` methods on static 7. **Numbers are JavaScript doubles end to end — raw-literal preservation is NOT a goal.** The Node service parsed every request body and upstream response with `JSON.parse` (doubles) and re-emitted with `JSON.stringify`, so integers above 2^53 have always rounded on these paths (e.g. `18446744073709551615` → `18446744073709552000`). `JsJson.Stringify` reproduces this byte-for-byte on purpose. Do not flag double-rounding on payload/response serialization as precision loss, and do not "fix" it by preserving raw literals — either direction breaks parity. The single deliberate exception is the Solana lamports raw-text scan (`PrivateApi.Chain`), which keeps `ToJsonString` because it extracts a u64 from raw text before any parse. 8. **Lone-surrogate `\u` escapes are valid input and output.** JavaScript strings are arbitrary UTF-16, so `JSON.parse`/`JSON.stringify` accept and re-emit lone surrogates; System.Text.Json throws on them in BOTH directions (string materialization AND the writer, regardless of encoder). All string extraction must go through `JsVal.TryGetStringLenient` and all tree serialization of client/upstream data through `JsJson.Stringify` — never raw `GetValue`/`TryGetValue` or `ToJsonString` on such data. Suggesting a switch back to the strict System.Text.Json APIs reintroduces production 500s. -## Hive RPC failover +## Upstream node failover -`HiveRpcClient` uses a per-node health tracker (adopted from the vision-next SDK): 429 responses park a node for `Retry-After` (or an escalating window), recent failures deprioritize it, and a latency EWMA orders the pool best-first with config order as tiebreak. RPC-level errors (JSON `error` field) surface immediately without failover — they're application errors, not node health. Covered by `HiveRpcFailoverTests`; keep new failover behavior test-backed. +`NodeHealthTracker` (adopted from the vision-next SDK) holds per-node health state: 429 responses park a node for `Retry-After` (or an escalating window), recent failures deprioritize it, and a latency EWMA orders the pool best-first with config order as tiebreak. Two clients build on it: + +- `HiveRpcClient` (Hive JSON-RPC): RPC-level errors (JSON `error` field) surface immediately without failover — they're application errors, not node health. +- `EngineRpcClient` (Hive-Engine): one instance per pool — the `/contracts` RPC pool and the history-API pool. The portfolio `Find` calls are fixed-shape queries that always yield a `result` array on a healthy node, so an error payload or non-JSON body *is* a node failure and rolls over to the next node. The raw passthroughs (`engine-api`, `engine-account-history`) fail over only on transport errors and 429/5xx; other responses belong to the caller's query and pipe as-is. + +Covered by `HiveRpcFailoverTests` and `EngineFailoverTests`; keep new failover behavior test-backed. ## Testing diff --git a/dotnet/EcencyApi.Tests/EngineFailoverTests.cs b/dotnet/EcencyApi.Tests/EngineFailoverTests.cs new file mode 100644 index 00000000..d50db264 --- /dev/null +++ b/dotnet/EcencyApi.Tests/EngineFailoverTests.cs @@ -0,0 +1,257 @@ +using System.Net; +using System.Text; +using System.Text.Json.Nodes; +using EcencyApi.Infrastructure; +using Xunit; + +namespace EcencyApi.Tests; + +/// +/// Exercises EngineRpcClient's health-aware failover against local stub HTTP +/// servers: a dead / error-payload / rate-limited node must roll over to the +/// next healthy node, the find calls must treat a missing result array as a +/// node failure, and the raw passthrough must preserve non-overload responses. +/// +public class EngineFailoverTests +{ + private const string ResultBody = + "{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":[{\"symbol\":\"LEO\",\"balance\":\"1.0\"}]}"; + + /// Minimal loopback HTTP server returning a scripted (status, body) per request. + private sealed class StubNode : IAsyncDisposable + { + private readonly HttpListener _listener = new(); + private readonly Func<(int Status, string Body)> _handler; + public string Url { get; } // no trailing slash: the client appends the path + public int Hits; + public volatile string? LastPathAndQuery; + + public StubNode(Func<(int, string)> handler) + { + _handler = handler; + var port = GetFreePort(); + Url = $"http://127.0.0.1:{port}"; + _listener.Prefixes.Add(Url + "/"); + _listener.Start(); + _ = Loop(); + } + + private async Task Loop() + { + while (_listener.IsListening) + { + HttpListenerContext ctx; + try { ctx = await _listener.GetContextAsync(); } + catch { return; } + + Interlocked.Increment(ref Hits); + LastPathAndQuery = ctx.Request.Url?.PathAndQuery; + var (status, bodyText) = _handler(); + var body = Encoding.UTF8.GetBytes(bodyText); + ctx.Response.StatusCode = status; + ctx.Response.ContentType = "application/json"; + ctx.Response.ContentLength64 = body.Length; + try + { + await ctx.Response.OutputStream.WriteAsync(body); + ctx.Response.Close(); + } + catch { /* client may have moved on */ } + } + } + + public static int GetFreePort() + { + var l = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0); + l.Start(); + var port = ((IPEndPoint)l.LocalEndpoint).Port; + l.Stop(); + return port; + } + + public async ValueTask DisposeAsync() + { + _listener.Stop(); + _listener.Close(); + await Task.CompletedTask; + } + } + + private static EngineRpcClient Client(params string[] nodes) => + new(nodes, Array.Empty>()); + + private static JsonObject FindPayload() => new() + { + ["jsonrpc"] = "2.0", + ["method"] = "find", + ["params"] = new JsonObject { ["contract"] = "tokens", ["table"] = "balances" }, + ["id"] = "1", + }; + + [Fact] + public async Task Find_DeadNode_FailsOverToHealthyNode() + { + var deadUrl = $"http://127.0.0.1:{StubNode.GetFreePort()}"; // nothing listening + await using var good = new StubNode(() => (200, ResultBody)); + + var result = await Client(deadUrl, good.Url).Find(FindPayload()); + + Assert.Single(result); + Assert.Equal("LEO", result[0]!["symbol"]!.GetValue()); + Assert.Equal(1, good.Hits); + } + + [Fact] + public async Task Find_ErrorPayload_IsANodeFailure_FailsOver() + { + // A healthy node always returns a result array for these fixed find + // queries; an error payload means the node is broken, not the query. + await using var broken = new StubNode(() => (200, "{\"error\":{\"message\":\"contract store unavailable\"}}")); + await using var good = new StubNode(() => (200, ResultBody)); + + var result = await Client(broken.Url, good.Url).Find(FindPayload()); + + Assert.Single(result); + Assert.Equal(1, broken.Hits); + Assert.Equal(1, good.Hits); + } + + [Fact] + public async Task Find_RateLimitedNode_IsParkedOnSubsequentCalls() + { + await using var limited = new StubNode(() => (429, "rate limited")); + await using var good = new StubNode(() => (200, ResultBody)); + + var client = Client(limited.Url, good.Url); + await client.Find(FindPayload()); + await client.Find(FindPayload()); + + // First call burns one hit discovering the 429; the park keeps the + // second call away entirely. + Assert.Equal(1, limited.Hits); + Assert.Equal(2, good.Hits); + } + + [Fact] + public async Task Find_FailedNode_IsDeprioritizedOnSubsequentCalls() + { + await using var bad = new StubNode(() => (503, "bad gateway")); + await using var good = new StubNode(() => (200, ResultBody)); + + var client = Client(bad.Url, good.Url); + await client.Find(FindPayload()); + var badAfterFirst = bad.Hits; + await client.Find(FindPayload()); + + Assert.Equal(badAfterFirst, bad.Hits); // recent failure: not retried + Assert.Equal(2, good.Hits); + } + + [Fact] + public async Task Find_AllNodesFailing_Throws() + { + await using var bad1 = new StubNode(() => (500, "boom")); + await using var bad2 = new StubNode(() => (200, "{\"error\":\"nope\"}")); + + await Assert.ThrowsAnyAsync(() => + Client(bad1.Url, bad2.Url).Find(FindPayload())); + Assert.Equal(1, bad1.Hits); + Assert.Equal(1, bad2.Hits); + } + + [Fact] + public async Task ContractsRaw_TransportFailure_FailsOver() + { + var deadUrl = $"http://127.0.0.1:{StubNode.GetFreePort()}"; + await using var good = new StubNode(() => (200, ResultBody)); + + var resp = await Client(deadUrl, good.Url).ContractsRaw(FindPayload()); + + Assert.Equal(200, resp.Status); + Assert.NotNull(resp.Json?["result"]); + } + + [Fact] + public async Task ContractsRaw_NonOverloadResponse_PipesAsIs_NoFailover() + { + // A 200 error payload (bad caller query) belongs to the caller — the + // passthrough must not shop it around the pool. + await using var first = new StubNode(() => (200, "{\"error\":\"unknown contract\"}")); + await using var second = new StubNode(() => (200, ResultBody)); + + var resp = await Client(first.Url, second.Url).ContractsRaw(FindPayload()); + + Assert.Equal(200, resp.Status); + Assert.Equal("unknown contract", resp.Json?["error"]?.GetValue()); + Assert.Equal(0, second.Hits); + } + + [Fact] + public async Task ContractsRaw_AllOverloaded_ReturnsLastUpstreamResponse() + { + await using var bad1 = new StubNode(() => (503, "overloaded")); + await using var bad2 = new StubNode(() => (502, "bad gateway")); + + var resp = await Client(bad1.Url, bad2.Url).ContractsRaw(FindPayload()); + + Assert.True(resp.Status is 502 or 503); + } + + [Fact] + public async Task ContractsRaw_RateLimitedNode_IsParkedOnSubsequentCalls() + { + await using var limited = new StubNode(() => (429, "rate limited")); + await using var good = new StubNode(() => (200, ResultBody)); + + var client = Client(limited.Url, good.Url); + var first = await client.ContractsRaw(FindPayload()); + var second = await client.ContractsRaw(FindPayload()); + + Assert.Equal(200, first.Status); + Assert.Equal(200, second.Status); + Assert.Equal(1, limited.Hits); // parked after the discovery hit + Assert.Equal(2, good.Hits); + } + + [Fact] + public void ParseRetryAfterMs_HugeDeltaSeconds_SaturatesInsteadOfOverflowing() + { + var ms = NodeHealthTracker.ParseRetryAfterMs("2147484"); // *1000 would overflow int + Assert.NotNull(ms); + Assert.True(ms > 0); + } + + [Fact] + public async Task GetRaw_DeadNode_FailsOver_PreservingPathAndQuery() + { + var deadUrl = $"http://127.0.0.1:{StubNode.GetFreePort()}"; + await using var good = new StubNode(() => (200, "[{\"symbol\":\"LEO\"}]")); + + var query = new[] + { + new KeyValuePair("account", "good-karma"), + new KeyValuePair("symbol", "LEO"), + }; + var resp = await Client(deadUrl, good.Url) + .GetRaw("/accountHistory", query, perAttemptTimeoutMs: 2000, maxAttempts: 2); + + Assert.Equal(200, resp.Status); + Assert.Equal("/accountHistory?account=good-karma&symbol=LEO", good.LastPathAndQuery); + } + + [Fact] + public async Task ContractsRaw_MaxAttempts_CapsPoolWalk() + { + await using var bad1 = new StubNode(() => (503, "x")); + await using var bad2 = new StubNode(() => (503, "x")); + await using var bad3 = new StubNode(() => (503, "x")); + await using var good = new StubNode(() => (200, ResultBody)); + + var resp = await Client(bad1.Url, bad2.Url, bad3.Url, good.Url) + .ContractsRaw(FindPayload(), maxAttempts: 3); + + // The cap stops the walk before the healthy 4th node; the last 503 is returned. + Assert.Equal(503, resp.Status); + Assert.Equal(0, good.Hits); + } +} diff --git a/dotnet/EcencyApi/Handlers/WalletApi.Engine.cs b/dotnet/EcencyApi/Handlers/WalletApi.Engine.cs index d8e53f36..01f8a26d 100644 --- a/dotnet/EcencyApi/Handlers/WalletApi.Engine.cs +++ b/dotnet/EcencyApi/Handlers/WalletApi.Engine.cs @@ -5,41 +5,18 @@ namespace EcencyApi.Handlers; /// -/// Port of wallet-api.ts lines 1748-2165: Hive-Engine node pool + failover, the -/// engine passthrough routes, the portfolio data fetchers, and the portfolio / -/// portfolioV2 orchestration (fail-fast leg timeouts, partial-degrade on error). +/// Port of wallet-api.ts lines 1748-2165: the engine passthrough routes, the +/// portfolio data fetchers, and the portfolio / portfolioV2 orchestration +/// (fail-fast leg timeouts, partial-degrade on error). The original module's +/// random-pick engine node rotation was replaced with EngineRpcClient's +/// health-aware failover — the random retry intermittently landed on dead +/// nodes twice and blanked engine data. /// public static partial class WalletApi { - private static readonly string[] EngineNodes = - { - "https://herpc.dtools.dev", - "https://api.hive-engine.com/rpc", - "https://ha.herpc.dtools.dev", - "https://herpc.kanibot.com", - "https://herpc.actifit.io", - }; - - // min and max included - private static int RandomIntFromInterval(int min, int max) => - (int)Math.Floor(Random.Shared.NextDouble() * (max - min + 1) + min); - - private static string PickRandomEngineUrl() => - $"{EngineNodes[RandomIntFromInterval(0, EngineNodes.Length - 1)]}/contracts"; - - // Module-level mutable base URL, reselected on engine failure (dhive-style - // rotation across the pool). Volatile: read/written from concurrent requests. - private static string _baseEngineUrl = PickRandomEngineUrl(); - private static string BaseEngineUrl - { - get => Volatile.Read(ref _baseEngineUrl); - set => Volatile.Write(ref _baseEngineUrl, value); - } - private const string BaseSpkUrl = "https://spk.good-karma.xyz"; private const string EngineRewardsUrl = "https://scot-api.hive-engine.com/"; private const string EngineChartUrl = "https://info-api.tribaldex.com/market/ohlcv"; - private const string EngineAccountHistoryUrl = "https://history.hive-engine.com/accountHistory"; private static readonly KeyValuePair[] EngineHeaders = { @@ -47,12 +24,33 @@ private static string BaseEngineUrl new("User-Agent", "Ecency"), }; + // Ordered by observed reliability (health-meter snapshot at last review); + // the tracker's latency EWMA reorders at runtime, config order is the + // cold-start order and tiebreak. + private static readonly EngineRpcClient EngineClient = new(new[] + { + "https://enginerpc.com", + "https://api.hive-engine.com/rpc", + "https://herpc.actifit.io", + "https://he.c0ff33a.uk", + "https://v6-he.atexoras.com:2083", + "https://api2.hive-engine.com/rpc", + "https://herpc.kanibot.com", + "https://herpc.dtools.dev", + }, EngineHeaders); + + private static readonly EngineRpcClient EngineHistoryClient = new(new[] + { + "https://history.hive-engine.com", + "https://v6-he.atexoras.com:8443", + }, EngineHeaders); + // ---- routes ---------------------------------------------------------- public static async Task Eapi(HttpContext ctx) { var data = await ctx.ReadBody(); - await Upstream.Pipe(EngineContractsRequest(data, BaseEngineUrl), ctx); + await Upstream.Pipe(EngineClient.ContractsRaw(data), ctx); } public static async Task Erewardapi(HttpContext ctx) @@ -72,8 +70,11 @@ await Upstream.Pipe( public static async Task EngineAccountHistory(HttpContext ctx) { var query = QueryOf(ctx); + // Per-attempt timeout keeps the original 30s envelope; a dead node + // fails fast (DNS/refused/5xx) so the second node still answers well + // within what callers already waited for. await Upstream.Pipe( - Upstream.BaseApiRequest(EngineAccountHistoryUrl, HttpMethod.Get, EngineHeaders, null, query, 30000), ctx); + EngineHistoryClient.GetRaw("/accountHistory", query, 15000, maxAttempts: 2), ctx); } private static List> QueryOf(HttpContext ctx) @@ -90,9 +91,6 @@ await Upstream.Pipe( // ---- raw engine calls ------------------------------------------------ - private static Task EngineContractsRequest(JsonNode? data, string url) => - Upstream.BaseApiRequest(url, HttpMethod.Post, EngineHeaders, data, null, 8000); - private static Task EngineRewardsRequest(string username, IEnumerable> query) { var url = $"{EngineRewardsUrl}/@{username}"; @@ -112,66 +110,17 @@ private static Task EngineRewardsRequest(string username, IEnu ["id"] = HiveEngine.IdOne, }; - private static JsonNode? ResultOf(UpstreamResponse r) => r.Json?["result"]; + public static Task FetchEngineBalances(string account) => + EngineClient.Find(EngineFindPayload(HiveEngine.ContractTokens, HiveEngine.TableBalances, + new JsonObject { ["account"] = account })); - public static async Task FetchEngineBalances(string account) - { - var data = EngineFindPayload(HiveEngine.ContractTokens, HiveEngine.TableBalances, - new JsonObject { ["account"] = account }); - try - { - var response = await EngineContractsRequest(data.DeepClone(), BaseEngineUrl); - var result = ResultOf(response); - if (result is not JsonArray a) throw new Exception("Failed to get engine balances"); - return a; - } - catch - { - BaseEngineUrl = PickRandomEngineUrl(); - var response = await EngineContractsRequest(data.DeepClone(), BaseEngineUrl); - var result = ResultOf(response); - if (result is not JsonArray a) throw new Exception("Failed to get engine balances"); - return a; - } - } + public static Task FetchEngineTokens(List tokens) => + EngineClient.Find(EngineFindPayload(HiveEngine.ContractTokens, HiveEngine.TableTokens, + new JsonObject { ["symbol"] = new JsonObject { ["$in"] = ToJsonArray(tokens) } })); - public static async Task FetchEngineTokens(List tokens) - { - var query = new JsonObject { ["symbol"] = new JsonObject { ["$in"] = ToJsonArray(tokens) } }; - var data = EngineFindPayload(HiveEngine.ContractTokens, HiveEngine.TableTokens, query); - try - { - var response = await EngineContractsRequest(data.DeepClone(), BaseEngineUrl); - if (ResultOf(response) is not JsonArray a) throw new Exception("Failed to get engine tokens data"); - return a; - } - catch - { - BaseEngineUrl = PickRandomEngineUrl(); - var response = await EngineContractsRequest(data.DeepClone(), BaseEngineUrl); - if (ResultOf(response) is not JsonArray a) throw new Exception("Failed to get engine tokens data"); - return a; - } - } - - public static async Task FetchEngineMetrics(List tokens) - { - var query = new JsonObject { ["symbol"] = new JsonObject { ["$in"] = ToJsonArray(tokens) } }; - var data = EngineFindPayload(HiveEngine.ContractMarket, HiveEngine.TableMetrics, query); - try - { - var response = await EngineContractsRequest(data.DeepClone(), BaseEngineUrl); - if (ResultOf(response) is not JsonArray a) throw new Exception("Failed to get engine metrics data"); - return a; - } - catch - { - BaseEngineUrl = PickRandomEngineUrl(); - var response = await EngineContractsRequest(data.DeepClone(), BaseEngineUrl); - if (ResultOf(response) is not JsonArray a) throw new Exception("Failed to get engine metrics data"); - return a; - } - } + public static Task FetchEngineMetrics(List tokens) => + EngineClient.Find(EngineFindPayload(HiveEngine.ContractMarket, HiveEngine.TableMetrics, + new JsonObject { ["symbol"] = new JsonObject { ["$in"] = ToJsonArray(tokens) } })); public static async Task FetchEngineRewards(string username) { diff --git a/dotnet/EcencyApi/Infrastructure/EngineRpcClient.cs b/dotnet/EcencyApi/Infrastructure/EngineRpcClient.cs new file mode 100644 index 00000000..883876dc --- /dev/null +++ b/dotnet/EcencyApi/Infrastructure/EngineRpcClient.cs @@ -0,0 +1,148 @@ +using System.Text.Json.Nodes; + +namespace EcencyApi.Infrastructure; + +/// +/// Hive-Engine upstream client with the same health-aware node failover as +/// HiveRpcClient (shared ). One instance per +/// pool of interchangeable nodes: the RPC (/contracts) pool and the history +/// API pool. Replaces the ported Node behavior — a sticky randomly-picked +/// node plus one blind random retry (and no retry at all for the history +/// host) — which intermittently blanked engine data whenever the pick landed +/// on a dead or flaky node. +/// +public sealed class EngineRpcClient +{ + private readonly string[] _nodes; + private readonly KeyValuePair[] _headers; + private readonly NodeHealthTracker _health; + + /// Base node URLs; "/contracts" is appended per call. + public EngineRpcClient(string[] nodes, KeyValuePair[] headers) + { + _nodes = nodes; + _headers = headers; + _health = new NodeHealthTracker(nodes.Length); + } + + private static long NowMs => Environment.TickCount64; + + /// + /// JSON-RPC find query requiring a "result" array in the response. The + /// portfolio queries are fixed shapes that always yield an array (possibly + /// empty) on a healthy node, so an error payload or non-JSON body is a node + /// problem, not an application error — mark it failed and try the next. + /// The per-attempt timeout is short because these calls run inside the + /// portfolioV2 leg budget; healthy engine nodes answer well under it. + /// + public async Task Find(JsonNode payload, int perAttemptTimeoutMs = 2000) + { + Exception? lastError = null; + + foreach (var nodeIndex in _health.OrderedNodeIndices()) + { + var node = _nodes[nodeIndex]; + var started = NowMs; + try + { + var resp = await Upstream.BaseApiRequest($"{node}/contracts", HttpMethod.Post, + _headers, payload, null, perAttemptTimeoutMs); + + if (resp.Status == 429) + { + _health.RecordRateLimited(nodeIndex, + NodeHealthTracker.ParseRetryAfterMs(resp.Headers.Get("Retry-After"))); + lastError = new Exception($"engine node {node} rate-limited (429)"); + continue; + } + if (resp.Status is < 200 or >= 300) + { + _health.RecordFailure(nodeIndex, NowMs - started); + lastError = new Exception($"engine node {node} returned {resp.Status}"); + continue; + } + if (resp.Json?["result"] is JsonArray result) + { + _health.RecordSuccess(nodeIndex, NowMs - started); + return result; + } + _health.RecordFailure(nodeIndex, NowMs - started); + lastError = new Exception($"engine node {node} returned no result array"); + } + catch (Exception e) // timeout, DNS failure, connection refused + { + _health.RecordFailure(nodeIndex, NowMs - started); + lastError = e; + } + } + + throw lastError ?? new InvalidOperationException("no engine nodes configured"); + } + + /// Raw /contracts passthrough for the engine-api proxy route. + public Task ContractsRaw(JsonNode? payload, + int perAttemptTimeoutMs = Upstream.DefaultTimeoutMs, int maxAttempts = 3) => + Passthrough(HttpMethod.Post, "/contracts", payload, null, perAttemptTimeoutMs, maxAttempts); + + /// GET passthrough (history API) with the caller's query forwarded. + public Task GetRaw(string path, + IEnumerable>? query, + int perAttemptTimeoutMs, int maxAttempts) => + Passthrough(HttpMethod.Get, path, null, query, perAttemptTimeoutMs, maxAttempts); + + /// + /// Proxy-route passthrough. Fails over only on transport errors and + /// overload statuses (429/5xx) — any other response belongs to the + /// caller's query and pipes as-is. If every tried node overloads, the last + /// upstream response is still returned so the client sees what the pool + /// said; a pool of pure transport failures rethrows for pipe()'s 504/500 + /// split. Attempts are capped: the wallet clients calling these routes + /// wait synchronously, and a pool-wide outage shouldn't multiply the full + /// per-attempt timeout by the pool size. + /// + private async Task Passthrough(HttpMethod method, string path, + JsonNode? payload, IEnumerable>? query, + int perAttemptTimeoutMs, int maxAttempts) + { + Exception? lastError = null; + UpstreamResponse? lastResponse = null; + var attempts = 0; + + foreach (var nodeIndex in _health.OrderedNodeIndices()) + { + if (attempts++ >= maxAttempts) break; + var node = _nodes[nodeIndex]; + var started = NowMs; + try + { + var resp = await Upstream.BaseApiRequest($"{node}{path}", method, + _headers, payload, query, perAttemptTimeoutMs); + + if (resp.Status == 429) + { + _health.RecordRateLimited(nodeIndex, + NodeHealthTracker.ParseRetryAfterMs(resp.Headers.Get("Retry-After"))); + lastResponse = resp; + continue; + } + if (resp.Status >= 500) + { + _health.RecordFailure(nodeIndex, NowMs - started); + lastResponse = resp; + continue; + } + + _health.RecordSuccess(nodeIndex, NowMs - started); + return resp; + } + catch (Exception e) + { + _health.RecordFailure(nodeIndex, NowMs - started); + lastError = e; + } + } + + if (lastResponse != null) return lastResponse; + throw lastError ?? new InvalidOperationException("no engine nodes configured"); + } +} diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index b8513932..57938bf3 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs @@ -1,60 +1,28 @@ -using System.Globalization; using System.Text; using System.Text.Json.Nodes; namespace EcencyApi.Infrastructure; /// -/// JSON-RPC client for Hive nodes with health-aware failover, adopting the -/// proven design of @ecency/sdk's NodeHealthTracker (simplified for a proxy -/// that makes a handful of call shapes): +/// JSON-RPC client for Hive nodes with health-aware failover. The per-node +/// health state and pool ordering live in +/// (shared with the Hive-Engine client); this class adds the Hive call +/// semantics: /// -/// - Per-node health state: consecutive failures, rate-limit parking, and a -/// latency EWMA used to order the pool best-first. -/// - 429 parks a node for the server's Retry-After when present, else an -/// escalating window (10s doubling to 60s); the escalation streak resets -/// after 120s without a throttle. Parked nodes sort last. -/// - A node with a recent failure (30s window) is deprioritized behind clean -/// nodes, so one bad response moves traffic away without banning the node. -/// - Healthy nodes are ordered by latency EWMA (alpha 0.3, trusted after 3 -/// samples, stale after 5 minutes); unproven nodes score a neutral prior -/// (1s) so an unknown node is explored before a proven-slow one. Config -/// order breaks ties, so cold start behaves exactly like the configured list. /// - RPC-level errors (JSON error field) surface immediately without failover /// (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. /// /// Not adopted (overkill at proxy call rates): request hedging, per-API failure /// profiles, and head-block staleness checks. /// public sealed class HiveRpcClient { - private const int RateLimitBaseMs = 10_000; - private const int RateLimitMaxMs = 60_000; - private const int RateLimitStreakResetMs = 120_000; - private const int RecentFailureWindowMs = 30_000; - private const double LatencyEwmaAlpha = 0.3; - private const int LatencyMinSamples = 3; - private const int LatencyMaxAgeMs = 5 * 60_000; - private const double LatencyUnprovenPriorMs = 1_000; - private const int SlowFailureFloorMs = 2_000; - - private sealed class NodeHealth - { - public int ConsecutiveFailures; - public long LastFailureAtMs; - public long RateLimitedUntilMs; - public int RateLimitStreak; - public long LastRateLimitAtMs; - public double? EwmaLatencyMs; - public int LatencySampleCount; - public long LatencyUpdatedAtMs; - } - private readonly string[] _nodes; private readonly int _timeoutMs; private readonly int _failoverThreshold; - private readonly NodeHealth[] _health; - private readonly object _lock = new(); + private readonly NodeHealthTracker _health; private int _seq; // timeoutMs 2000 / failoverThreshold 2 mirror the dhive Client options the @@ -64,11 +32,7 @@ public HiveRpcClient(string[] nodes, int timeoutMs = 2000, int failoverThreshold _nodes = nodes; _timeoutMs = timeoutMs; _failoverThreshold = Math.Max(1, failoverThreshold); - _health = new NodeHealth[nodes.Length]; - for (var i = 0; i < nodes.Length; i++) - { - _health[i] = new NodeHealth(); - } + _health = new NodeHealthTracker(nodes.Length); } public sealed class RpcException : Exception @@ -78,105 +42,6 @@ public RpcException(string message) : base(message) { } private static long NowMs => Environment.TickCount64; - // ---- health bookkeeping (lock-guarded; contention is negligible) ------ - - private void RecordSuccess(int nodeIndex, double elapsedMs) - { - lock (_lock) - { - var h = _health[nodeIndex]; - h.ConsecutiveFailures = 0; - h.RateLimitStreak = 0; - RecordLatency(h, elapsedMs); - } - } - - private void RecordFailure(int nodeIndex, double elapsedMs) - { - lock (_lock) - { - var h = _health[nodeIndex]; - 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) - { - RecordLatency(h, elapsedMs); - } - } - } - - private void RecordRateLimited(int nodeIndex, int? retryAfterMs) - { - lock (_lock) - { - var h = _health[nodeIndex]; - var now = NowMs; - h.ConsecutiveFailures++; - h.LastFailureAtMs = now; - if (now - h.LastRateLimitAtMs > RateLimitStreakResetMs) - { - h.RateLimitStreak = 0; - } - var parkMs = retryAfterMs - ?? Math.Min(RateLimitBaseMs << Math.Min(h.RateLimitStreak, 3), RateLimitMaxMs); - h.RateLimitedUntilMs = now + parkMs; - h.RateLimitStreak++; - h.LastRateLimitAtMs = now; - } - } - - private static void RecordLatency(NodeHealth h, double elapsedMs) - { - var now = NowMs; - // A stale profile restarts from scratch so an idle process re-learns - // instead of ranking on old data. - if (h.LatencyUpdatedAtMs > 0 && now - h.LatencyUpdatedAtMs > LatencyMaxAgeMs) - { - h.EwmaLatencyMs = null; - h.LatencySampleCount = 0; - } - h.EwmaLatencyMs = h.EwmaLatencyMs is { } prev - ? LatencyEwmaAlpha * elapsedMs + (1 - LatencyEwmaAlpha) * prev - : elapsedMs; - h.LatencySampleCount++; - h.LatencyUpdatedAtMs = now; - } - - /// - /// 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. - /// - private List OrderedNodeIndices() - { - lock (_lock) - { - var now = NowMs; - return Enumerable.Range(0, _nodes.Length) - .Select(i => - { - var h = _health[i]; - var parked = h.RateLimitedUntilMs > 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); - }) - .OrderBy(x => x.Parked) - .ThenBy(x => x.RecentFailure) - .ThenBy(x => x.Score) - .ThenBy(x => x.Index) - .Select(x => x.Index) - .ToList(); - } - } - // ---- calls ------------------------------------------------------------- public async Task Call(string api, string method, JsonNode @params) @@ -194,7 +59,7 @@ private List OrderedNodeIndices() Exception? lastError = null; - foreach (var nodeIndex in OrderedNodeIndices()) + foreach (var nodeIndex in _health.OrderedNodeIndices()) { var node = _nodes[nodeIndex]; @@ -204,14 +69,14 @@ private List OrderedNodeIndices() try { var result = await CallNode(node, body); - RecordSuccess(nodeIndex, NowMs - started); + _health.RecordSuccess(nodeIndex, NowMs - started); return result; } catch (RpcException) { // The node answered; the error is the application's. No // failover (dhive semantics), and no failure mark. - RecordSuccess(nodeIndex, NowMs - started); + _health.RecordSuccess(nodeIndex, NowMs - started); throw; } catch (NodeUnavailableException e) @@ -219,11 +84,11 @@ private List OrderedNodeIndices() lastError = e; if (e.IsRateLimit) { - RecordRateLimited(nodeIndex, e.RetryAfterMs); + _health.RecordRateLimited(nodeIndex, e.RetryAfterMs); } else { - RecordFailure(nodeIndex, NowMs - started); + _health.RecordFailure(nodeIndex, NowMs - started); } if (e.AdvanceImmediately) { @@ -233,7 +98,7 @@ private List OrderedNodeIndices() catch (Exception e) { lastError = e; - RecordFailure(nodeIndex, NowMs - started); + _health.RecordFailure(nodeIndex, NowMs - started); } } } @@ -267,24 +132,6 @@ public NodeUnavailableException(string message, bool advanceImmediately, // skip it now rather than burning a retry that will fail the same way. private static bool IsOverloadStatus(int status) => status is 429 or 502 or 503 or 504; - /// Retry-After: delta-seconds or an HTTP date (RFC 9110). - private static int? ParseRetryAfterMs(string? header) - { - if (string.IsNullOrWhiteSpace(header)) return null; - var t = header.Trim(); - if (int.TryParse(t, NumberStyles.None, CultureInfo.InvariantCulture, out var seconds)) - { - return seconds >= 0 ? seconds * 1000 : null; - } - if (DateTimeOffset.TryParse(t, CultureInfo.InvariantCulture, - DateTimeStyles.AssumeUniversal | DateTimeStyles.AdjustToUniversal, out var when)) - { - var ms = (when - DateTimeOffset.UtcNow).TotalMilliseconds; - return ms > 0 ? (int)Math.Min(ms, int.MaxValue) : 0; - } - return null; - } - private async Task CallNode(string node, string body) { using var req = new HttpRequestMessage(HttpMethod.Post, node) @@ -313,7 +160,7 @@ public NodeUnavailableException(string message, bool advanceImmediately, { var status = (int)resp.StatusCode; var retryAfter = status == 429 - ? ParseRetryAfterMs(resp.Headers.TryGetValues("Retry-After", out var vals) + ? NodeHealthTracker.ParseRetryAfterMs(resp.Headers.TryGetValues("Retry-After", out var vals) ? vals.FirstOrDefault() : null) : null; diff --git a/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs b/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs new file mode 100644 index 00000000..f4450e46 --- /dev/null +++ b/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs @@ -0,0 +1,178 @@ +using System.Globalization; + +namespace EcencyApi.Infrastructure; + +/// +/// Per-node health bookkeeping shared by the upstream RPC clients (Hive and +/// Hive-Engine), adopting the proven design of @ecency/sdk's NodeHealthTracker +/// (simplified for a proxy that makes a handful of call shapes): +/// +/// - Per-node health state: consecutive failures, rate-limit parking, and a +/// latency EWMA used to order the pool best-first. +/// - 429 parks a node for the server's Retry-After when present, else an +/// escalating window (10s doubling to 60s); the escalation streak resets +/// after 120s without a throttle. Parked nodes sort last. +/// - A node with a recent failure (30s window) is deprioritized behind clean +/// nodes, so one bad response moves traffic away without banning the node. +/// - Healthy nodes are ordered by latency EWMA (alpha 0.3, trusted after 3 +/// samples, stale after 5 minutes); unproven nodes score a neutral prior +/// (1s) so an unknown node is explored before a proven-slow one. Config +/// order breaks ties, so cold start behaves exactly like the configured list. +/// +public sealed class NodeHealthTracker +{ + private const int RateLimitBaseMs = 10_000; + private const int RateLimitMaxMs = 60_000; + private const int RateLimitStreakResetMs = 120_000; + private const int RecentFailureWindowMs = 30_000; + private const double LatencyEwmaAlpha = 0.3; + private const int LatencyMinSamples = 3; + private const int LatencyMaxAgeMs = 5 * 60_000; + private const double LatencyUnprovenPriorMs = 1_000; + private const int SlowFailureFloorMs = 2_000; + + private sealed class NodeHealth + { + public int ConsecutiveFailures; + public long LastFailureAtMs; + public long RateLimitedUntilMs; + public int RateLimitStreak; + public long LastRateLimitAtMs; + public double? EwmaLatencyMs; + public int LatencySampleCount; + public long LatencyUpdatedAtMs; + } + + private readonly NodeHealth[] _health; + private readonly object _lock = new(); + + public NodeHealthTracker(int nodeCount) + { + _health = new NodeHealth[nodeCount]; + for (var i = 0; i < nodeCount; i++) + { + _health[i] = new NodeHealth(); + } + } + + private static long NowMs => Environment.TickCount64; + + // ---- health bookkeeping (lock-guarded; contention is negligible) ------ + + public void RecordSuccess(int nodeIndex, double elapsedMs) + { + lock (_lock) + { + var h = _health[nodeIndex]; + h.ConsecutiveFailures = 0; + h.RateLimitStreak = 0; + RecordLatency(h, elapsedMs); + } + } + + public void RecordFailure(int nodeIndex, double elapsedMs) + { + lock (_lock) + { + var h = _health[nodeIndex]; + 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) + { + RecordLatency(h, elapsedMs); + } + } + } + + public void RecordRateLimited(int nodeIndex, int? retryAfterMs) + { + lock (_lock) + { + var h = _health[nodeIndex]; + var now = NowMs; + h.ConsecutiveFailures++; + h.LastFailureAtMs = now; + if (now - h.LastRateLimitAtMs > RateLimitStreakResetMs) + { + h.RateLimitStreak = 0; + } + var parkMs = retryAfterMs + ?? Math.Min(RateLimitBaseMs << Math.Min(h.RateLimitStreak, 3), RateLimitMaxMs); + h.RateLimitedUntilMs = now + parkMs; + h.RateLimitStreak++; + h.LastRateLimitAtMs = now; + } + } + + private static void RecordLatency(NodeHealth h, double elapsedMs) + { + var now = NowMs; + // A stale profile restarts from scratch so an idle process re-learns + // instead of ranking on old data. + if (h.LatencyUpdatedAtMs > 0 && now - h.LatencyUpdatedAtMs > LatencyMaxAgeMs) + { + h.EwmaLatencyMs = null; + h.LatencySampleCount = 0; + } + h.EwmaLatencyMs = h.EwmaLatencyMs is { } prev + ? LatencyEwmaAlpha * elapsedMs + (1 - LatencyEwmaAlpha) * prev + : elapsedMs; + h.LatencySampleCount++; + h.LatencyUpdatedAtMs = now; + } + + /// + /// 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. + /// + public List OrderedNodeIndices() + { + lock (_lock) + { + var now = NowMs; + return Enumerable.Range(0, _health.Length) + .Select(i => + { + var h = _health[i]; + var parked = h.RateLimitedUntilMs > 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); + }) + .OrderBy(x => x.Parked) + .ThenBy(x => x.RecentFailure) + .ThenBy(x => x.Score) + .ThenBy(x => x.Index) + .Select(x => x.Index) + .ToList(); + } + } + + /// Retry-After: delta-seconds or an HTTP date (RFC 9110). + public static int? ParseRetryAfterMs(string? header) + { + if (string.IsNullOrWhiteSpace(header)) return null; + var t = header.Trim(); + if (long.TryParse(t, NumberStyles.None, CultureInfo.InvariantCulture, out var seconds)) + { + // Clamp before converting: a huge delta-seconds must saturate, not + // overflow into a negative park window. + return seconds >= 0 ? (int)(Math.Min(seconds, int.MaxValue / 1000L) * 1000) : null; + } + if (DateTimeOffset.TryParse(t, CultureInfo.InvariantCulture, + DateTimeStyles.AssumeUniversal | DateTimeStyles.AdjustToUniversal, out var when)) + { + var ms = (when - DateTimeOffset.UtcNow).TotalMilliseconds; + return ms > 0 ? (int)Math.Min(ms, int.MaxValue) : 0; + } + return null; + } +}