rpc pool: rank nodes per call class, not on one blended latency EWMA

The tracker kept one latency EWMA per node across every call shape, so the
ranking was learned from the calls that dominate by count (point reads) and then
used to pick a node for feed-shaped queries, whose cost is several times higher
and varies by an order of magnitude between nodes.

Latency is now kept per (node, call class). Health stays node-wide: a node that
is not answering is not answering for any class. The SSR cache classifies every
allowlisted method and the three feed-shaped reads are heavy;
SSR_RPC_CALL_CLASSES=0 collapses it back to one profile without a rebuild.

Stats gain per-node heavy_ewma_ms/heavy_samples plus a per-method class.

Author: feruzm

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.

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
 

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<int> _handler; // returns HTTP status; 200 => valid RPC result
+        private readonly Func<string, int> _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<int> handler)
+        public StubNode(Func<int> handler) : this(_ => handler())
+        {
+        }
+
+        /// <param name="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.</param>
+        public StubNode(Func<string, int> 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()
             }
         }
 
+        /// <summary>The qualified method of a JSON-RPC request, in either the
+        /// dotted form or the legacy `call` envelope.</summary>
+        private static string MethodOf(string body)
+        {
+            try
+            {
+                var req = JsonNode.Parse(body);
+                var method = req?["method"]?.GetValue<string>() ?? "";
+                return method == "call"
+                    ? (req?["params"]?[0]?.GetValue<string>() ?? "") + "." +
+                      (req?["params"]?[1]?.GetValue<string>() ?? "")
+                    : 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<int>());
+        Assert.Equal(3, view["heavy_samples"]!.GetValue<int>());
+        Assert.True(view["ewma_ms"]!.GetValue<double>() < 1000);
+        Assert.True(view["heavy_ewma_ms"]!.GetValue<double>() > 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<int>());
+        Assert.Equal(0, view["heavy_samples"]!.GetValue<int>());
+        Assert.Null(view["heavy_ewma_ms"]);
+    }
+
     [Fact]
     public async Task MalformedResultNode_FailsOverWithoutRetry()
     {

dotnet/EcencyApi.Tests/NodeCallClassTests.cs

@@ -0,0 +1,146 @@
+using EcencyApi.Infrastructure;
+using Xunit;
+
+namespace EcencyApi.Tests;
+
+/// <summary>
+/// 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.
+/// </summary>
+public class NodeCallClassTests
+{
+    private static (NodeHealthTracker Tracker, Action<long> 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 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);
+    }
+}

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,51 @@ 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<int>());
+        Assert.Equal(1, view["heavy_samples"]!.GetValue<int>());
+    }
+
+    [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<int>());
+        Assert.Equal(0, view["heavy_samples"]!.GetValue<int>());
+    }
 }

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;
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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);