diff --git a/CLAUDE.md b/CLAUDE.md index 271d1b2f..69eda89a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -68,7 +68,7 @@ Handlers are `public static async Task Name(HttpContext ctx)` methods on static ## Upstream node failover -`NodeHealthTracker` (adopted from the vision-web 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: +`NodeHealthTracker` (adopted from the vision-web 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. That EWMA is kept per **call class** (`CallClass.Cheap` / `CallClass.Heavy`) because upstream cost is bimodal: a point read costs a fraction of a feed-shaped query, while which node is quickest differs between the two. A caller says which class a call belongs to and the pool is ordered from that class's profile. Everything about *whether* a node is answering (consecutive failures, failure parking, rate-limit parking, half-open admission) stays node-wide. 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. The typed helpers additionally validate the result *shape* (`get_accounts` → array, `get_dynamic_global_properties` → object): a 200 with valid JSON but no usable result is a node failure that fails over — without this, a node serving malformed 200s is recorded as healthy and stays ranked first (observed in production as multi-hour windows of token-validation 401s). - `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. diff --git a/README.md b/README.md index 2598026a..3ef0e10c 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,7 @@ docker run -it --rm -p 4000:4000 \ | `SSR_RPC_NODE_TIMEOUT_MS` | per-node timeout of the SSR RPC cache's own client, one attempt per node (default `1200`) | | `SSR_RPC_NODES` | comma-separated node pool for that client (default: the shared pool) | | `SSR_RPC_MAX_FILLS` / `SSR_RPC_MAX_QUEUED_FILLS` | bound on upstream fills in progress (default `64`) and on fills waiting for that bound (default `256`); beyond the latter a miss fails fast | +| `SSR_RPC_CALL_CLASSES` | order the node pool from a per-call-class latency profile (default on); `0`, `false` or `off` files every read under one profile | ## Swarm diff --git a/dotnet/EcencyApi.Tests/HiveRpcFailoverTests.cs b/dotnet/EcencyApi.Tests/HiveRpcFailoverTests.cs index 4b0d38e3..bcb80281 100644 --- a/dotnet/EcencyApi.Tests/HiveRpcFailoverTests.cs +++ b/dotnet/EcencyApi.Tests/HiveRpcFailoverTests.cs @@ -17,7 +17,7 @@ public class HiveRpcFailoverTests private sealed class StubNode : IAsyncDisposable { private readonly HttpListener _listener = new(); - private readonly Func _handler; // returns HTTP status; 200 => valid RPC result + private readonly Func _handler; // returns HTTP status; 200 => valid RPC result public string Url { get; } public int Hits; @@ -31,7 +31,14 @@ private sealed class StubNode : IAsyncDisposable "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":[{\"name\":\"served-by\",\"port\":\"" + Url + "\",\"posting_json_metadata\":" + (ServesMetadata ? "\"{\\\"profile\\\":{}}\"" : "\"\"") + "}]}"; - public StubNode(Func handler) + public StubNode(Func handler) : this(_ => handler()) + { + } + + /// Given the qualified method of the request, returns + /// the scripted status. Lets one node answer point reads quickly and feed + /// queries slowly, which is the shape the call-class split exists for. + public StubNode(Func handler) { _handler = handler; var port = GetFreePort(); @@ -50,7 +57,12 @@ private async Task Loop() catch { return; } Interlocked.Increment(ref Hits); - var status = _handler(); + string requestBody; + using (var reader = new StreamReader(ctx.Request.InputStream)) + { + requestBody = await reader.ReadToEndAsync(); + } + var status = _handler(MethodOf(requestBody)); byte[] body; if (status == 200) { @@ -113,6 +125,25 @@ private async Task Loop() } } + /// The qualified method of a JSON-RPC request, in either the + /// dotted form or the legacy `call` envelope. + private static string MethodOf(string body) + { + try + { + var req = JsonNode.Parse(body); + var method = req?["method"]?.GetValue() ?? ""; + return method == "call" + ? (req?["params"]?[0]?.GetValue() ?? "") + "." + + (req?["params"]?[1]?.GetValue() ?? "") + : method; + } + catch + { + return ""; + } + } + private static int GetFreePort() { var l = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0); @@ -472,6 +503,65 @@ public async Task ProvenSlowNode_IsDemotedByLatencyEwma() Assert.True(fast.Hits >= 1); } + // The call-class split, end to end: upstream cost is bimodal, so a node can be + // the right choice for point reads and the wrong one for feed queries. With a + // single latency profile per node the ranking is learned from whichever class + // dominates by count and then used to pick a node for the other. + [Fact] + public async Task ANodeSlowOnlyOnFeedQueries_KeepsThePointReadsAndLosesTheFeeds() + { + // -2 answers 200 after 1.5s, above the 1s unproven prior; 200 is immediate. + await using var mixed = new StubNode(m => m.StartsWith("bridge.", StringComparison.Ordinal) ? -2 : 200); + await using var spare = new StubNode(_ => 200); + + var client = new HiveRpcClient(new[] { mixed.Url, spare.Url }, timeoutMs: 5000, failoverThreshold: 1); + + // Both classes start unproven, so config order sends them to the first node. + for (var i = 0; i < 3; i++) + { + await client.Call("condenser_api", "get_accounts", new JsonArray(), callClass: CallClass.Cheap); + await client.CallMethod("bridge.get_ranked_posts", new JsonObject(), callClass: CallClass.Heavy); + } + Assert.Equal(6, mixed.Hits); + Assert.Equal(0, spare.Hits); + + // Its heavy profile is now trusted and above the prior, so the next feed + // query explores the node nothing is known about... + await client.CallMethod("bridge.get_ranked_posts", new JsonObject(), callClass: CallClass.Heavy); + Assert.Equal(1, spare.Hits); + + // ...while point reads stay where they are measured to be quick. + await client.Call("condenser_api", "get_accounts", new JsonArray(), callClass: CallClass.Cheap); + Assert.Equal(7, mixed.Hits); + Assert.Equal(1, spare.Hits); + + // Same node, two profiles, learned from their own samples only. + var view = client.HealthSnapshot()[0]!; + Assert.Equal(4, view["samples"]!.GetValue()); + Assert.Equal(3, view["heavy_samples"]!.GetValue()); + Assert.True(view["ewma_ms"]!.GetValue() < 1000); + Assert.True(view["heavy_ewma_ms"]!.GetValue() > 1000); + } + + [Fact] + public async Task ACallerThatMakesOnePointReadShape_LeavesTheHeavyProfileEmpty() + { + // The default class: a client whose calls are all one shape keeps exactly + // one profile per node, as it did before classes existed. + await using var only = new StubNode(() => 200); + + var client = new HiveRpcClient(new[] { only.Url }, timeoutMs: 1500); + for (var i = 0; i < 3; i++) + { + await client.Call("condenser_api", "get_accounts", new JsonArray()); + } + + var view = client.HealthSnapshot()[0]!; + Assert.Equal(3, view["samples"]!.GetValue()); + Assert.Equal(0, view["heavy_samples"]!.GetValue()); + Assert.Null(view["heavy_ewma_ms"]); + } + [Fact] public async Task MalformedResultNode_FailsOverWithoutRetry() { diff --git a/dotnet/EcencyApi.Tests/NodeCallClassTests.cs b/dotnet/EcencyApi.Tests/NodeCallClassTests.cs new file mode 100644 index 00000000..0895ce77 --- /dev/null +++ b/dotnet/EcencyApi.Tests/NodeCallClassTests.cs @@ -0,0 +1,157 @@ +using EcencyApi.Infrastructure; +using Xunit; + +namespace EcencyApi.Tests; + +/// +/// The health tracker's per-call-class latency, driven directly with an +/// injected clock: latency is the only thing that splits by class, everything +/// that decides whether a node is answering at all stays node-wide. +/// +public class NodeCallClassTests +{ + private static (NodeHealthTracker Tracker, Action Advance) Build(int nodes) + { + long now = 0; + return (new NodeHealthTracker(nodes, () => now), ms => now += ms); + } + + private static double? Ewma(NodeHealthTracker t, int node, CallClass cls) => + t.Snapshot()[node].Latency.First(l => l.Class == cls).EwmaMs; + + private static int Samples(NodeHealthTracker t, int node, CallClass cls) => + t.Snapshot()[node].Latency.First(l => l.Class == cls).Samples; + + [Fact] + public void CallClassValues_StayContiguousFromZero() + { + // Each node holds one latency profile per class in an array indexed by the + // enum value, on the ordering hot path. A gap, a negative member or a + // renumbering would index outside that array, so the layout is pinned here + // instead of trusted, and the check costs nothing. + var values = Enum.GetValues().Select(v => (int)v).ToArray(); + Assert.Equal(Enumerable.Range(0, values.Length).ToArray(), values); + } + + [Fact] + public void EachClassKeepsItsOwnLatencyProfile() + { + var (t, _) = Build(1); + for (var i = 0; i < 3; i++) t.RecordSuccess(0, 100, CallClass.Cheap); + for (var i = 0; i < 3; i++) t.RecordSuccess(0, 1500, CallClass.Heavy); + + Assert.Equal(100, Ewma(t, 0, CallClass.Cheap)); + Assert.Equal(1500, Ewma(t, 0, CallClass.Heavy)); + Assert.Equal(3, Samples(t, 0, CallClass.Cheap)); + Assert.Equal(3, Samples(t, 0, CallClass.Heavy)); + } + + [Fact] + public void ANodeQuickOnPointReadsAndSlowOnFeedQueries_LeadsOnlyTheCheapOrdering() + { + // The whole point of the split: node 0 wins the cheap ranking on its own + // measurements and must NOT carry that win into the heavy ranking, where + // it is slower than a node nothing is known about. + var (t, _) = Build(2); + for (var i = 0; i < 3; i++) + { + t.RecordSuccess(0, 100, CallClass.Cheap); + t.RecordSuccess(0, 1500, CallClass.Heavy); + } + + Assert.Equal(new[] { 0, 1 }, t.OrderedNodeIndices(CallClass.Cheap)); + Assert.Equal(new[] { 1, 0 }, t.OrderedNodeIndices(CallClass.Heavy)); + } + + [Fact] + public void OneClassGoingStale_LeavesTheOtherProfileAlone() + { + // Staleness is per class. A class the traffic has moved away from + // becoming unproven again is exploration, not a penalty: the node keeps + // its other profile and all of its health. + var (t, advance) = Build(2); + advance(1_000); // a profile stamped at tick 0 reads as never stamped + for (var i = 0; i < 3; i++) + { + t.RecordSuccess(0, 100, CallClass.Cheap); + t.RecordSuccess(0, 1500, CallClass.Heavy); + } + + advance(6 * 60_000); + t.RecordSuccess(0, 120, CallClass.Cheap); + + Assert.Equal(1, Samples(t, 0, CallClass.Cheap)); // reset and re-learning + Assert.Equal(120, Ewma(t, 0, CallClass.Cheap)); + Assert.Equal(3, Samples(t, 0, CallClass.Heavy)); // untouched + Assert.Equal(1500, Ewma(t, 0, CallClass.Heavy)); + // ...but stale, so it no longer orders anything: node 0 scores the prior + // for heavy, so config order breaks the tie with the untried node. + Assert.Equal(new[] { 0, 1 }, t.OrderedNodeIndices(CallClass.Heavy)); + } + + [Fact] + public void ATimeoutIsALatencySample_ForTheClassThatTimedOut() + { + // Floored above the unproven prior so a node that never answers a heavy + // query cannot outrank nodes never tried for one. The cheap profile learns + // nothing from it, because nothing cheap was measured. + var (t, _) = Build(2); + for (var i = 0; i < 3; i++) t.RecordFailure(0, 300, CallClass.Heavy, timedOut: true); + + Assert.True(Ewma(t, 0, CallClass.Heavy) > 1000); + Assert.Equal(3, Samples(t, 0, CallClass.Heavy)); + Assert.Null(Ewma(t, 0, CallClass.Cheap)); + Assert.Equal(0, Samples(t, 0, CallClass.Cheap)); + } + + [Fact] + public void AFailureParkedNode_IsSkippedForEveryClass() + { + // "Not answering" is not a per-class property: a parked node is out of + // both orderings while any other node can take the call. + var (t, _) = Build(2); + for (var i = 0; i < 3; i++) t.RecordFailure(0, 300, CallClass.Heavy, timedOut: true); + + Assert.Equal(new[] { 1 }, t.OrderedNodeIndices(CallClass.Heavy)); + Assert.Equal(new[] { 1 }, t.OrderedNodeIndices(CallClass.Cheap)); + } + + [Fact] + public void ARateLimitedNode_SortsLastForEveryClass() + { + var (t, _) = Build(2); + for (var i = 0; i < 3; i++) t.RecordSuccess(0, 10, CallClass.Cheap); + for (var i = 0; i < 3; i++) t.RecordSuccess(0, 10, CallClass.Heavy); + t.RecordRateLimited(0, 5_000); + + Assert.Equal(new[] { 1, 0 }, t.OrderedNodeIndices(CallClass.Cheap)); + Assert.Equal(new[] { 1, 0 }, t.OrderedNodeIndices(CallClass.Heavy)); + } + + [Fact] + public void ARecentFailureOnOneClass_DemotesTheNodeForBoth() + { + // Deliberate. It is also the narrow scope of the split: only the latency + // score is per class. A node that just failed is a node that just failed, + // whatever the call was, so it sorts behind clean nodes for everything. + var (t, _) = Build(2); + for (var i = 0; i < 3; i++) t.RecordSuccess(0, 10, CallClass.Cheap); + t.RecordFailure(0, 50, CallClass.Heavy); + + Assert.Equal(new[] { 1, 0 }, t.OrderedNodeIndices(CallClass.Cheap)); + Assert.Equal(new[] { 1, 0 }, t.OrderedNodeIndices(CallClass.Heavy)); + } + + [Fact] + public void ASuccessOnOneClass_ClearsNodeWideFailureState() + { + var (t, _) = Build(2); + for (var i = 0; i < 3; i++) t.RecordFailure(0, 300, CallClass.Heavy, timedOut: true); + Assert.Equal(new[] { 1 }, t.OrderedNodeIndices(CallClass.Cheap)); + + t.RecordSuccess(0, 20, CallClass.Cheap); + + Assert.Equal(2, t.OrderedNodeIndices(CallClass.Cheap).Count); + Assert.Equal(0, t.Snapshot()[0].FailureParkedForMs); + } +} diff --git a/dotnet/EcencyApi.Tests/SsrRpcTests.cs b/dotnet/EcencyApi.Tests/SsrRpcTests.cs index 239f1371..7741ca6e 100644 --- a/dotnet/EcencyApi.Tests/SsrRpcTests.cs +++ b/dotnet/EcencyApi.Tests/SsrRpcTests.cs @@ -95,6 +95,7 @@ public async ValueTask DisposeAsync() } private static readonly SsrRpc.MethodPolicy Post = SsrRpc.Allowlist["bridge.get_post"]; + private static readonly SsrRpc.MethodPolicy Ranked = SsrRpc.Allowlist["bridge.get_ranked_posts"]; private static readonly SsrRpc.MethodPolicy Props = SsrRpc.Allowlist["condenser_api.get_dynamic_global_properties"]; private static void Use(RpcStub stub, long cacheBytes = 1 << 20, int budgetMs = 1500, int maxFills = 64, int maxQueued = 256) @@ -106,6 +107,7 @@ private static void Use(RpcStub stub, long cacheBytes = 1 << 20, int budgetMs = SsrRpc.MaxQueuedFills = maxQueued; SsrRpc.SecretDigest = null; SsrRpc.Now = () => Environment.TickCount64; + SsrRpc.CallClasses = true; SsrRpc.ResetForTests(); } @@ -630,4 +632,85 @@ 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")); } + + [Fact] + public void Allowlist_classifies_the_feed_shaped_reads_as_heavy() + { + // A page of feed rows or a whole comment tree, built per request by + // hivemind; the rest are point reads. The pool is ordered from the + // latency profile of the class, so a misclassified method is ranked on + // measurements of a different call shape. + var heavy = SsrRpc.Allowlist.Values + .Where(p => p.Class == CallClass.Heavy) + .Select(p => p.Key) + .OrderBy(k => k, StringComparer.Ordinal) + .ToArray(); + Assert.Equal( + new[] { "bridge.get_account_posts", "bridge.get_discussion", "bridge.get_ranked_posts" }, + heavy); + } + + [Fact] + public async Task A_heavy_read_is_measured_in_the_heavy_profile() + { + await using var stub = new RpcStub(); + Use(stub); + + await SsrRpc.Resolve(Ranked, new JsonObject { ["sort"] = "trending", ["tag"] = "" }); + await SsrRpc.Resolve(Post, P("a", "b")); + + var view = SsrRpc.Client.HealthSnapshot()[0]!; + Assert.Equal(1, view["samples"]!.GetValue()); + Assert.Equal(1, view["heavy_samples"]!.GetValue()); + } + + [Fact] + public async Task Stats_report_the_call_class_of_each_method_and_the_heavy_node_profile() + { + // The stats route is the only way to tell, on a running deployment, + // whether a node's blended number was hiding feed-query cost, so the + // fields that answer that have to be in the payload, not just in the + // snapshot the route reads from. + await using var stub = new RpcStub(); + Use(stub); + SsrRpc.SecretDigest = SsrRpc.Digest("right-secret"); + try + { + await SsrRpc.Resolve(Ranked, new JsonObject { ["sort"] = "trending", ["tag"] = "" }); + await SsrRpc.Resolve(Post, P("a", "b")); + + var stats = Request("GET", "/private-api/ssr/stats", "right-secret"); + await SsrRpc.Stats(stats); + var body = JsonNode.Parse(ResponseText(stats))!; + + Assert.True(body["call_classes"]!.GetValue()); + Assert.Equal("heavy", body["methods"]!["bridge.get_ranked_posts"]!["class"]!.GetValue()); + Assert.Equal("cheap", body["methods"]!["bridge.get_post"]!["class"]!.GetValue()); + + var node = Assert.Single(body["nodes"]!.AsArray())!; + Assert.Equal(1, node["samples"]!.GetValue()); + Assert.Equal(1, node["heavy_samples"]!.GetValue()); + Assert.NotNull(node["heavy_ewma_ms"]); + } + finally + { + SsrRpc.SecretDigest = null; + } + } + + [Fact] + public async Task Call_classes_off_files_every_read_under_the_cheap_profile() + { + // The kill switch restores the single-profile ordering this service had + // before call classes existed, without a rebuild. + await using var stub = new RpcStub(); + Use(stub); + SsrRpc.CallClasses = false; + + await SsrRpc.Resolve(Ranked, new JsonObject { ["sort"] = "trending", ["tag"] = "" }); + + var view = SsrRpc.Client.HealthSnapshot()[0]!; + Assert.Equal(1, view["samples"]!.GetValue()); + Assert.Equal(0, view["heavy_samples"]!.GetValue()); + } } diff --git a/dotnet/EcencyApi/Config.cs b/dotnet/EcencyApi/Config.cs index 46f04a43..e0d2e586 100644 --- a/dotnet/EcencyApi/Config.cs +++ b/dotnet/EcencyApi/Config.cs @@ -65,6 +65,15 @@ public static class Config public static int SsrMaxQueuedFills { get; } = int.TryParse(Env("SSR_RPC_MAX_QUEUED_FILLS"), out var q) && q > 0 ? q : 256; + /// + /// Whether the cache orders the node pool per call class (see CallClass) or + /// files every read under one profile, which is how it behaved before classes + /// existed. On by default; set to 0 to collapse it on a running deployment + /// without a rebuild. Break-glass, so the off spellings are permissive. + /// + public static bool SsrCallClasses { get; } = + Env("SSR_RPC_CALL_CLASSES")?.Trim().ToLowerInvariant() is not ("0" or "false" or "off"); + private static string? NonEmpty(string? value) => string.IsNullOrWhiteSpace(value) ? null : value.Trim(); private static string? Env(string name) => Environment.GetEnvironmentVariable(name); diff --git a/dotnet/EcencyApi/Handlers/SsrRpc.cs b/dotnet/EcencyApi/Handlers/SsrRpc.cs index 5b2c91aa..506d57d9 100644 --- a/dotnet/EcencyApi/Handlers/SsrRpc.cs +++ b/dotnet/EcencyApi/Handlers/SsrRpc.cs @@ -32,7 +32,13 @@ namespace EcencyApi.Handlers; /// public static partial class SsrRpc { - internal sealed record MethodPolicy(string Api, string Method, int TtlMs) + /// What this read costs upstream. Feed-shaped reads (a + /// ranked page, an account's posts, a comment tree) cost several times a point + /// read. Unlike a point read the cost also varies by an order of magnitude + /// 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) { public string Key => $"{Api}.{Method}"; } @@ -42,18 +48,23 @@ internal sealed record MethodPolicy(string Api, string Method, int TtlMs) // votes/payout within tens of seconds, a profile or community rarely. internal static readonly IReadOnlyDictionary Allowlist = new[] { - new MethodPolicy("bridge", "get_ranked_posts", 15_000), - new MethodPolicy("bridge", "get_account_posts", 30_000), - new MethodPolicy("bridge", "get_post", 30_000), - new MethodPolicy("bridge", "get_discussion", 30_000), - new MethodPolicy("bridge", "get_profile", 60_000), - new MethodPolicy("bridge", "get_profiles", 60_000), - new MethodPolicy("bridge", "get_community", 300_000), - new MethodPolicy("bridge", "list_communities", 300_000), - new MethodPolicy("condenser_api", "get_accounts", 30_000), - new MethodPolicy("condenser_api", "get_content", 30_000), - new MethodPolicy("condenser_api", "get_dynamic_global_properties", 3_000), - new MethodPolicy("condenser_api", "get_trending_tags", 300_000), + // Heavy: a page of feed rows or a whole comment tree, built per request by + // hivemind. Cheap: a point read of one post, account, profile or community. + // The class is a per-method proxy for cost: the same method can be cheaper + // or dearer depending on its params, so it follows the shape of the read + // rather than any one call. + new MethodPolicy("bridge", "get_ranked_posts", 15_000, CallClass.Heavy), + 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), + 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), + new MethodPolicy("bridge", "list_communities", 300_000, CallClass.Cheap), + new MethodPolicy("condenser_api", "get_accounts", 30_000, CallClass.Cheap), + new MethodPolicy("condenser_api", "get_content", 30_000, CallClass.Cheap), + new MethodPolicy("condenser_api", "get_dynamic_global_properties", 3_000, CallClass.Cheap), + new MethodPolicy("condenser_api", "get_trending_tags", 300_000, CallClass.Cheap), }.ToDictionary(p => p.Key, p => p); internal const string HeaderName = "X-Ecency-Internal"; @@ -77,6 +88,10 @@ internal sealed record MethodPolicy(string Api, string Method, int TtlMs) internal static int BudgetMs = Config.SsrBudgetMs; + // Off collapses every read onto the cheap profile, which is how the pool was + // ordered before call classes existed. Replaceable for tests. + internal static bool CallClasses = Config.SsrCallClasses; + // Clock behind the lookup deadline and the attach/expiry timestamps; // replaceable so tests can drive the post-deadline paths deterministically. internal static Func Now = () => Environment.TickCount64; @@ -361,7 +376,8 @@ private static async Task Fill(MethodPolicy policy, JsonNode @params, string key // "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()); + result = await Client.CallMethod($"{policy.Api}.{policy.Method}", @params.DeepClone(), + callClass: CallClasses ? policy.Class : CallClass.Cheap); } finally { @@ -453,6 +469,13 @@ public static async Task Stats(HttpContext ctx) var c = kv.Value; methods[kv.Key] = new JsonObject { + // Which latency profile orders the pool for this method, so the + // timeout/slow_fill columns can be read per class. Null only if a + // counter outlives its allowlist entry, which the route cannot + // produce (every counter key comes from a policy). + ["class"] = Allowlist.TryGetValue(kv.Key, out var p) + ? p.Class.ToString().ToLowerInvariant() + : null, ["hit"] = Interlocked.Read(ref c.Hit), ["miss"] = Interlocked.Read(ref c.Miss), ["coalesced"] = Interlocked.Read(ref c.Coalesced), @@ -472,6 +495,7 @@ public static async Task Stats(HttpContext ctx) ["budget"] = Cache.Budget, }, ["budget_ms"] = BudgetMs, + ["call_classes"] = CallClasses, ["methods"] = methods, ["nodes"] = Client.HealthSnapshot(), }); diff --git a/dotnet/EcencyApi/Infrastructure/EngineRpcClient.cs b/dotnet/EcencyApi/Infrastructure/EngineRpcClient.cs index 883876dc..0529bf83 100644 --- a/dotnet/EcencyApi/Infrastructure/EngineRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/EngineRpcClient.cs @@ -15,6 +15,11 @@ public sealed class EngineRpcClient { private readonly string[] _nodes; private readonly KeyValuePair[] _headers; + // Every call this client makes is filed under one latency class. The tracker + // splits latency per call class for pools whose call mix is bimodal; this one + // is not split. (Its /contracts pool does mix a fixed-shape Find with the + // caller-supplied raw passthrough, which is a candidate for the same + // treatment. Out of scope here.) private readonly NodeHealthTracker _health; /// Base node URLs; "/contracts" is appended per call. @@ -39,7 +44,7 @@ public async Task Find(JsonNode payload, int perAttemptTimeoutMs = 20 { Exception? lastError = null; - foreach (var nodeIndex in _health.OrderedNodeIndices()) + foreach (var nodeIndex in _health.OrderedNodeIndices(CallClass.Cheap)) { var node = _nodes[nodeIndex]; var started = NowMs; @@ -57,21 +62,21 @@ public async Task Find(JsonNode payload, int perAttemptTimeoutMs = 20 } if (resp.Status is < 200 or >= 300) { - _health.RecordFailure(nodeIndex, NowMs - started); + _health.RecordFailure(nodeIndex, NowMs - started, CallClass.Cheap); lastError = new Exception($"engine node {node} returned {resp.Status}"); continue; } if (resp.Json?["result"] is JsonArray result) { - _health.RecordSuccess(nodeIndex, NowMs - started); + _health.RecordSuccess(nodeIndex, NowMs - started, CallClass.Cheap); return result; } - _health.RecordFailure(nodeIndex, NowMs - started); + _health.RecordFailure(nodeIndex, NowMs - started, CallClass.Cheap); lastError = new Exception($"engine node {node} returned no result array"); } catch (Exception e) // timeout, DNS failure, connection refused { - _health.RecordFailure(nodeIndex, NowMs - started); + _health.RecordFailure(nodeIndex, NowMs - started, CallClass.Cheap); lastError = e; } } @@ -108,7 +113,7 @@ private async Task Passthrough(HttpMethod method, string path, UpstreamResponse? lastResponse = null; var attempts = 0; - foreach (var nodeIndex in _health.OrderedNodeIndices()) + foreach (var nodeIndex in _health.OrderedNodeIndices(CallClass.Cheap)) { if (attempts++ >= maxAttempts) break; var node = _nodes[nodeIndex]; @@ -127,17 +132,17 @@ private async Task Passthrough(HttpMethod method, string path, } if (resp.Status >= 500) { - _health.RecordFailure(nodeIndex, NowMs - started); + _health.RecordFailure(nodeIndex, NowMs - started, CallClass.Cheap); lastResponse = resp; continue; } - _health.RecordSuccess(nodeIndex, NowMs - started); + _health.RecordSuccess(nodeIndex, NowMs - started, CallClass.Cheap); return resp; } catch (Exception e) { - _health.RecordFailure(nodeIndex, NowMs - started); + _health.RecordFailure(nodeIndex, NowMs - started, CallClass.Cheap); lastError = e; } } diff --git a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs index 55ab4ba3..9e887682 100644 --- a/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs +++ b/dotnet/EcencyApi/Infrastructure/HiveRpcClient.cs @@ -14,6 +14,10 @@ namespace EcencyApi.Infrastructure; /// - Overload statuses (429/502/503/504) advance to the next node immediately /// instead of burning a same-node retry. /// +/// Latency is tracked per call class (see ) because the +/// reads made through this client are bimodal in cost. Failure state is not: it +/// stays node-wide. +/// /// Not adopted (overkill at proxy call rates): request hedging, per-API failure /// profiles, and head-block staleness checks. /// @@ -53,7 +57,7 @@ public JsonArray HealthSnapshot() var arr = new JsonArray(); foreach (var v in _health.Snapshot()) { - arr.Add(new JsonObject + var node = new JsonObject { ["node"] = Uri.TryCreate(_nodes[v.Index], UriKind.Absolute, out var u) ? u.Host : _nodes[v.Index], ["calls"] = v.Calls, @@ -61,14 +65,24 @@ public JsonArray HealthSnapshot() ["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, - ["failure_rate"] = Math.Round(v.FailureRate, 3), - }); + }; + // One EWMA per call class (see CallClass): the pool is ordered from + // the profile of the class being called, so both must be readable to + // tell "this node is slow" from "this node is slow at feed queries". + // ewma_ms/samples stay the cheap class, which is what they have always + // reported in practice, since cheap calls dominate by count. + var cheap = v.Latency.First(l => l.Class == CallClass.Cheap); + var heavy = v.Latency.First(l => l.Class == CallClass.Heavy); + node["ewma_ms"] = cheap.EwmaMs is { } ce ? JsonValue.Create(Math.Round(ce, 1)) : null; + node["samples"] = cheap.Samples; + node["heavy_ewma_ms"] = heavy.EwmaMs is { } he ? JsonValue.Create(Math.Round(he, 1)) : null; + node["heavy_samples"] = heavy.Samples; + node["consecutive_failures"] = v.ConsecutiveFailures; + node["recent_failure"] = v.RecentFailure; + node["rate_limited_for_ms"] = v.RateLimitedForMs; + node["parked_for_ms"] = v.FailureParkedForMs; + node["failure_rate"] = Math.Round(v.FailureRate, 3); + arr.Add(node); } return arr; } @@ -104,9 +118,14 @@ public RpcException(string message) : base(message) { } /// valid 200 with a usable array, so shape validation passes and the latency EWMA /// keeps such a node ranked first — silently blanking every metadata-derived /// feature (portfolio engine/chain token visibility) with no error and no log. + /// Which latency profile this call's timings belong + /// to. That profile is also the one the pool is ordered from for this call. + /// Defaults to Cheap, so a caller that makes one shape of call keeps exactly + /// one profile per node. public Task Call(string api, string method, JsonNode @params, Func? validateResult = null, - Func? preferResult = null) + Func? preferResult = null, + CallClass callClass = CallClass.Cheap) { // The legacy `call` envelope the Node service always sent. hived resolves // it for its own APIs (condenser_api, database_api); hivemind's `bridge` @@ -118,7 +137,7 @@ public RpcException(string message) : base(message) { } ["method"] = "call", ["params"] = new JsonArray(api, method, @params), }; - return Send(request, method, validateResult, preferResult); + return Send(request, method, validateResult, preferResult, callClass); } /// @@ -128,7 +147,8 @@ public RpcException(string message) : base(message) { } /// public Task CallMethod(string qualifiedMethod, JsonNode @params, Func? validateResult = null, - Func? preferResult = null) + Func? preferResult = null, + CallClass callClass = CallClass.Cheap) { var request = new JsonObject { @@ -137,12 +157,13 @@ public RpcException(string message) : base(message) { } ["method"] = qualifiedMethod, ["params"] = @params, }; - return Send(request, qualifiedMethod, validateResult, preferResult); + return Send(request, qualifiedMethod, validateResult, preferResult, callClass); } private async Task Send(JsonObject request, string method, Func? validateResult, - Func? preferResult) + Func? preferResult, + CallClass callClass) { // JsJson: a lone-surrogate username from a client token must serialize // (JSON.stringify semantics) instead of throwing in the writer. @@ -153,7 +174,7 @@ public RpcException(string message) : base(message) { } var haveUnpreferred = false; var unpreferredCount = 0; - foreach (var nodeIndex in _health.OrderedNodeIndices()) + foreach (var nodeIndex in _health.OrderedNodeIndices(callClass)) { var node = _nodes[nodeIndex]; @@ -175,7 +196,7 @@ public RpcException(string message) : base(message) { } } // The node is healthy either way — record the success before // deciding whether its answer is the one we wanted. - _health.RecordSuccess(nodeIndex, NowMs - started); + _health.RecordSuccess(nodeIndex, NowMs - started, callClass); if (preferResult != null && !preferResult(result)) { // Keep the first such answer as the floor and try the next @@ -196,7 +217,7 @@ public RpcException(string message) : base(message) { } { // The node answered; the error is the application's. No // failover (dhive semantics), and no failure mark. - _health.RecordSuccess(nodeIndex, NowMs - started); + _health.RecordSuccess(nodeIndex, NowMs - started, callClass); // ...but if we only came to this node to improve on an answer we // already hold, its error belongs to the optional probe, not to // the caller's request. Rethrowing here would fail a call that @@ -211,7 +232,7 @@ public RpcException(string message) : base(message) { } { _health.RecordRateLimited(nodeIndex, e.RetryAfterMs); } - else if (_health.RecordFailure(nodeIndex, NowMs - started, e.IsTimeout)) + else if (_health.RecordFailure(nodeIndex, NowMs - started, callClass, e.IsTimeout)) { break; // this failure parked the node: no same-node retry } @@ -223,7 +244,7 @@ public RpcException(string message) : base(message) { } catch (Exception e) { lastError = e; - if (_health.RecordFailure(nodeIndex, NowMs - started)) + if (_health.RecordFailure(nodeIndex, NowMs - started, callClass)) { break; } diff --git a/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs b/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs index b984b998..2d963efd 100644 --- a/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs +++ b/dotnet/EcencyApi/Infrastructure/NodeHealthTracker.cs @@ -2,6 +2,33 @@ namespace EcencyApi.Infrastructure; +/// +/// How expensive one upstream call is, for latency accounting only. +/// +/// Cost is bimodal across the reads this service makes: point reads answer in a +/// few hundred milliseconds from nearly every node, while feed-shaped queries +/// cost several times that and vary by an order of magnitude BETWEEN nodes. With +/// a single latency profile per node the ranking is learned from whichever class +/// dominates by count (the cheap one) and then used to pick a node for the other. +/// That is how a node quick on point reads and slow on feed queries ends up +/// leading the pool for feed queries too. +/// +/// Only *how fast* a node is differs by class. Whether it is answering at all +/// (consecutive failures, rate-limit parking, failure parking, half-open +/// admission) stays node-wide: a node that is not answering is not answering for +/// any class. +/// +public enum CallClass +{ + /// Point reads. The default, so a client that makes one shape of + /// call keeps exactly one profile per node, as before. + Cheap = 0, + + /// Feed-shaped queries: seconds where a point read takes + /// milliseconds. Node-dependent in a way point reads are not. + Heavy = 1, +} + /// /// Per-node health bookkeeping shared by the upstream RPC clients (Hive and /// Hive-Engine), adopting the proven design of @ecency/sdk's NodeHealthTracker @@ -14,10 +41,11 @@ namespace EcencyApi.Infrastructure; /// 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. +/// - Healthy nodes are ordered by the latency EWMA of the call class being +/// ordered (alpha 0.3, trusted after 3 samples, stale after 5 minutes); a +/// node that class has not proven scores 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 { @@ -28,6 +56,18 @@ public sealed class NodeHealthTracker private const double LatencyEwmaAlpha = 0.3; private const int LatencyMinSamples = 3; private const int LatencyMaxAgeMs = 5 * 60_000; + // Score for a node whose profile for the class being ordered is not trusted + // yet, so an unexplored node is tried before a proven-slow one. ONE prior for + // both classes, deliberately: + // - It must stay below the caller's per-node timeout, or it stops separating + // anything. Every latency a client can observe is bounded by that timeout, + // so a prior above it is never exceeded by a real measurement: the first + // node to reach LatencyMinSamples would outrank every untried node forever + // and no other node would ever be sampled. + // - The alternative of scoring an unproven class from the node's OTHER class + // is worse than it looks: the two are on different scales (a heavy query + // costs several times a point read on the same node), so every + // heavy-unproven node would outrank every heavy-proven one. 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 @@ -48,6 +88,18 @@ public sealed class NodeHealthTracker private const double FailureRateAlpha = 0.1; private const double FailureRateParkFloor = 0.5; + // The class is an index into each node's latency array, so CallClass values + // must stay contiguous from zero. + private static readonly int CallClassCount = Enum.GetValues().Length; + + /// One node's measured latency for one call class. + private sealed class LatencyProfile + { + public double? EwmaMs; + public int SampleCount; + public long UpdatedAtMs; + } + private sealed class NodeHealth { public int ConsecutiveFailures; @@ -65,17 +117,21 @@ private sealed class NodeHealth // 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; + // Latency is the one thing kept per call class; see CallClass. + public readonly LatencyProfile[] Latency = + Enumerable.Range(0, CallClassCount).Select(_ => new LatencyProfile()).ToArray(); // Lifetime counters, for the stats endpoint. public long Calls, Successes, Failures, Timeouts, RateLimits; } + /// One node's latency for one call class, as reported by + /// ; one entry per , in enum order. + public sealed record ClassLatencyView(CallClass Class, double? EwmaMs, int Samples); + /// 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, + IReadOnlyList Latency, int ConsecutiveFailures, bool RecentFailure, long RateLimitedForMs, long FailureParkedForMs, double FailureRate); private readonly NodeHealth[] _health; @@ -96,7 +152,10 @@ public NodeHealthTracker(int nodeCount, Func? clock = null) // ---- health bookkeeping (lock-guarded; contention is negligible) ------ - public void RecordSuccess(int nodeIndex, double elapsedMs) + /// Which latency profile the sample belongs to. + /// Required, not defaulted: a call site that forgets it would silently file a + /// heavy measurement under the cheap profile and still compile and pass. + public void RecordSuccess(int nodeIndex, double elapsedMs, CallClass callClass) { lock (_lock) { @@ -113,7 +172,7 @@ public void RecordSuccess(int nodeIndex, double elapsedMs) // not be pushed out again by a stale deadline the moment another // node's park lapses. h.FailureParkedUntilMs = 0; - RecordLatency(h, elapsedMs); + RecordLatency(h, callClass, elapsedMs); } } @@ -123,7 +182,9 @@ public void RecordSuccess(int nodeIndex, double elapsedMs) /// 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) + /// Which latency profile the sample belongs to. The + /// failure itself is node-wide health either way. + public bool RecordFailure(int nodeIndex, double elapsedMs, CallClass callClass, bool timedOut = false) { lock (_lock) { @@ -141,11 +202,15 @@ public bool RecordFailure(int nodeIndex, double elapsedMs, bool timedOut = false // 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)); + RecordLatency(h, callClass, Math.Max(elapsedMs, LatencyUnprovenPriorMs + 1)); } + // Only a slow failure is a latency statement; an instant refusal says + // the node is down, not that it is slow. Under a per-node timeout + // below this floor the branch cannot be reached at all, so every + // failure sample that class records is the censored timeout above. else if (elapsedMs >= SlowFailureFloorMs) { - RecordLatency(h, elapsedMs); + RecordLatency(h, callClass, elapsedMs); } var notAnswering = h.Successes == 0 || h.FailureRate >= FailureRateParkFloor; if (h.ConsecutiveHardFailures >= FailureParkThreshold && notAnswering) @@ -180,21 +245,24 @@ public void RecordRateLimited(int nodeIndex, int? retryAfterMs) } } - private void RecordLatency(NodeHealth h, double elapsedMs) + private void RecordLatency(NodeHealth h, CallClass callClass, double elapsedMs) { var now = NowMs; + var p = h.Latency[(int)callClass]; // 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) + // instead of ranking on old data. Per class: a class that has not been + // called in a while is unproven again, which is exploration, not a + // penalty: the node keeps its other profile and all of its health. + if (p.UpdatedAtMs > 0 && now - p.UpdatedAtMs > LatencyMaxAgeMs) { - h.EwmaLatencyMs = null; - h.LatencySampleCount = 0; + p.EwmaMs = null; + p.SampleCount = 0; } - h.EwmaLatencyMs = h.EwmaLatencyMs is { } prev + p.EwmaMs = p.EwmaMs is { } prev ? LatencyEwmaAlpha * elapsedMs + (1 - LatencyEwmaAlpha) * prev : elapsedMs; - h.LatencySampleCount++; - h.LatencyUpdatedAtMs = now; + p.SampleCount++; + p.UpdatedAtMs = now; } /// @@ -205,8 +273,12 @@ private void RecordLatency(NodeHealth h, double elapsedMs) /// 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". + /// + /// Only the latency score depends on ; every + /// tier above it is node-wide, so the two classes agree on which nodes are + /// usable at all and differ only in the order of the usable ones. /// - public List OrderedNodeIndices() + public List OrderedNodeIndices(CallClass callClass) { lock (_lock) { @@ -219,10 +291,11 @@ public List OrderedNodeIndices() 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; + var p = h.Latency[(int)callClass]; + var latencyUsable = p.EwmaMs is not null + && p.SampleCount >= LatencyMinSamples + && now - p.UpdatedAtMs <= LatencyMaxAgeMs; + var score = latencyUsable ? p.EwmaMs!.Value : LatencyUnprovenPriorMs; return (Index: i, Parked: parked, Dead: dead, RecentFailure: recentFailure, Score: score); }) .ToList(); @@ -293,8 +366,11 @@ public List Snapshot() return Enumerable.Range(0, _health.Length).Select(i => { var h = _health[i]; + var latency = Enumerable.Range(0, CallClassCount) + .Select(c => new ClassLatencyView((CallClass)c, h.Latency[c].EwmaMs, h.Latency[c].SampleCount)) + .ToList(); return new NodeView(i, h.Calls, h.Successes, h.Failures, h.Timeouts, h.RateLimits, - h.EwmaLatencyMs, h.LatencySampleCount, h.ConsecutiveFailures, + latency, h.ConsecutiveFailures, h.ConsecutiveFailures > 0 && now - h.LastFailureAtMs < RecentFailureWindowMs, Math.Max(0, h.RateLimitedUntilMs - now), Math.Max(0, h.FailureParkedUntilMs - now), h.FailureRate);