SSR RPC cache: rank nodes per call class, not on a single latency EWMA

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,157 @@
+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 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<CallClass>().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);
+    }
+}