Improve integration test resilience with rate-limiting adjustments fo…

CHANGELOG.md

@@ -61,13 +61,15 @@ version (on the earlier 0.x releases, minor versions could carry them).
   that invokes .NET asserts the SDK resolver's actual selected version, so preinstalled newer SDKs cannot
   invalidate the minimum-SDK gate. Release publishing now fails unless the pushed tag exactly matches the
   package version, requires private live-cluster endpoints, and admits no skipped or inconclusive unit,
-  read, or streaming paths. Faucet-dependent devnet writes run serially through a one-request-per-second
-  limiter as a separate probe: deterministic failures still block publication, while classified faucet/rate-
-  limit failures are reported as inconclusive instead of making release availability depend on the shared
-  faucet. The probe verifies the canonical devnet genesis hash before any write, and Native-AOT publishes and
-  runs the exact packed artifact from an isolated package cache before pushing it. A duplicate immutable NuGet
-  version succeeds only when the repository-signed NuGet copy proves it contains the exact pre-staged canonical
-  package; mismatched or unverifiable duplicates fail visibly.
+  read, or streaming paths. Live HTTP reads and devnet writes use test-only two-request-per-second limiters,
+  while WebSocket probes run serially with paced starts so low-tier provider quotas do not turn a sequential
+  suite into a burst. Faucet-dependent devnet writes remain a separate probe: deterministic failures still
+  block publication, while classified faucet/rate-limit failures are reported as inconclusive instead of
+  making release availability depend on the shared faucet. The probe verifies the canonical devnet genesis
+  hash before any write, and Native-AOT publishes and runs the exact packed artifact from an isolated package
+  cache before pushing it. A duplicate immutable NuGet version succeeds only when the repository-signed NuGet
+  copy proves it contains the exact pre-staged canonical package; mismatched or unverifiable duplicates fail
+  visibly.
 - Added centrally configured StyleCop analysis to every project. Rider, Roslyn, and StyleCop now share an
   explicit modifier order, while repository-conflicting documentation and legacy-layout rules are suppressed
   in `.editorconfig` instead of producing misleading IDE warnings. CI and release now require a clean

CLAUDE.md

@@ -79,7 +79,7 @@ SolSharp/
 - `IDE1006` is disabled for `tests/**` so `Method_Scenario_Expectation` names are allowed.
 - For constructor-throws-only tests use an explicit discard: `Action act = () => _ = new T(...);`.
 - **Arrange / Act / Assert comments.** Mark the three phases with `// Arrange`, `// Act`, `// Assert`. When the call under test and its check are a single fluent statement (exception delegates, `(await …).Should()…`), use one `// Act & Assert`. Skip the labels on expression-bodied or single-statement `[TestCase]` tests where there is nothing to separate — never restructure a test body just to fit them.
-- **Integration tests** live in `SolSharp.IntegrationTests`, hit a real cluster, and run as part of `dotnet test`. They are tagged `[Category("Integration")]`; read/streaming tests default to public mainnet (`SOLSHARP_RPC_URL` / `SOLSHARP_WS_URL` override), and the write suite (airdrop, transfer, durable nonce) always targets devnet (`SOLSHARP_DEVNET_RPC_URL` override) — never mainnet. Write fixtures additionally carry `[Category("DevnetWrite")]`, are non-parallel, and use the existing resilience pipeline with a one-request-per-second token bucket. No key is ever committed. Ordinary runs report transient endpoint/faucet failures as inconclusive. The release gate is strict for unit/read/streaming tests, while it attempts the faucet-dependent write probe separately so a shared-faucet 429 cannot block publication; deterministic write-path failures still fail. Skip all live tests for a fast offline run with `dotnet test --filter "TestCategory!=Integration"`.
+- **Integration tests** live in `SolSharp.IntegrationTests`, hit a real cluster, and run as part of `dotnet test`. They are tagged `[Category("Integration")]`; read/streaming tests default to public mainnet (`SOLSHARP_RPC_URL` / `SOLSHARP_WS_URL` override), and the write suite (airdrop, transfer, durable nonce) always targets devnet (`SOLSHARP_DEVNET_RPC_URL` override) — never mainnet. HTTP read and write harnesses use shared two-request-per-second token buckets; WebSocket probes are serialized and their starts are paced at 500 ms. Write fixtures additionally carry `[Category("DevnetWrite")]` and are non-parallel. No key is ever committed. Ordinary runs report transient endpoint/faucet failures as inconclusive. The release gate is strict for unit/read/streaming tests, while it attempts the faucet-dependent write probe separately so a shared-faucet 429 cannot block publication; deterministic write-path failures still fail. Skip all live tests for a fast offline run with `dotnet test --filter "TestCategory!=Integration"`.
 
 ## Security (money-critical)
 

README.md

@@ -337,9 +337,9 @@ live cluster, plus a write suite (airdrop, transfer, durable nonce) that always
 default to the public mainnet endpoint (`SOLSHARP_RPC_URL` / `SOLSHARP_WS_URL` override); the write suite uses
 the public devnet endpoint (`SOLSHARP_DEVNET_RPC_URL` override); no credentials are committed. These tests hit
 the network, so they tolerate rate limits by reporting inconclusive rather than failing, and are tagged
-`Integration`. The devnet write harness is serialized and throttled to one RPC request per second, but the
-shared faucet can still reject `requestAirdrop` independently of RPC traffic. For a fast, offline-only run,
-exclude them:
+`Integration`. Live HTTP reads and devnet writes use two-request-per-second test-only limiters; WebSocket
+probes run serially with starts spaced by 500 ms. The shared faucet can still reject `requestAirdrop`
+independently of RPC traffic. For a fast, offline-only run, exclude them:
 
 ```bash
 dotnet test --filter "TestCategory!=Integration"

src/SolSharp.Rpc/Streaming/SolanaWsClient.cs

@@ -1619,7 +1619,7 @@ private async Task RouteAsync(string message, ConnectionEpoch epoch)
                 if (errorElement.ValueKind != JsonValueKind.Object ||
                     !errorElement.TryGetProperty("code", out var codeElement) ||
                     codeElement.ValueKind != JsonValueKind.Number ||
-                    !codeElement.TryGetInt64(out _) ||
+                    !codeElement.TryGetInt32(out _) ||
                     !errorElement.TryGetProperty("message", out var messageElement) ||
                     messageElement.ValueKind != JsonValueKind.String)
                 {
@@ -1779,16 +1779,11 @@ await SendUnsubscribeReservedAsync(
 
     private void CompletePendingError(int requestId, ConnectionEpoch epoch, JsonElement errorElement)
     {
-        var detail = errorElement.ValueKind == JsonValueKind.Object &&
-                     errorElement.TryGetProperty("message", out var errorMessage) &&
-                     errorMessage.ValueKind == JsonValueKind.String
-            ? errorMessage.GetString()
-            : errorElement.GetRawText();
-        var code = errorElement.ValueKind == JsonValueKind.Object &&
-                   errorElement.TryGetProperty("code", out var codeElement) &&
-                   codeElement.TryGetInt64(out var codeValue)
-            ? codeValue
-            : 0;
+        var detail = errorElement.GetProperty("message").GetString()!;
+        var code = errorElement.GetProperty("code").GetInt32();
+        JsonElement? data = errorElement.TryGetProperty("data", out var dataElement)
+            ? dataElement
+            : null;
 
         string method;
         lock (_stateGate)
@@ -1810,7 +1805,8 @@ private void CompletePendingError(int requestId, ConnectionEpoch epoch, JsonElem
             {
                 pending.Acked.TrySetException(
                     new InvalidOperationException(
-                        $"The node rejected '{method}' (code {code}): {detail}"));
+                        $"The node rejected '{method}' (code {code}): {detail}",
+                        new RpcException(code, detail, data)));
             }
         }
 

tests/SolSharp.IntegrationTests/DevnetWriteIntegrationTests.cs

@@ -22,7 +22,7 @@ public static class DevnetWriteIntegrationTests
         TokenLimit = 1,
         QueueLimit = 128,
         QueueProcessingOrder = QueueProcessingOrder.OldestFirst,
-        ReplenishmentPeriod = TimeSpan.FromSeconds(1),
+        ReplenishmentPeriod = TimeSpan.FromMilliseconds(500),
         TokensPerPeriod = 1,
         AutoReplenishment = true,
     });

tests/SolSharp.IntegrationTests/IntegrationEnvironmentTests.cs

@@ -37,6 +37,20 @@ public void TransientRpcFailure_ReturnsTrue(int code)
             // Assert
             result.Should().BeTrue();
         }
+
+        [TestCase(-32007, true)]
+        [TestCase(-32602, false)]
+        public void WrappedRpcFailure_UsesInnerRpcClassification(int code, bool expected)
+        {
+            // Arrange
+            var exception = new InvalidOperationException("WebSocket subscription rejected", new RpcException(code, "RPC"));
+
+            // Act
+            var result = IntegrationEnvironment.IsTransient(exception);
+
+            // Assert
+            result.Should().Be(expected);
+        }
     }
 
     [TestFixture]

tests/SolSharp.IntegrationTests/RpcReadIntegrationTests.cs

@@ -1,3 +1,4 @@
+using System.Threading.RateLimiting;
 using FluentAssertions;
 using Microsoft.Extensions.DependencyInjection;
 using NUnit.Framework;
@@ -15,6 +16,16 @@ namespace SolSharp.IntegrationTests;
 /// </summary>
 public static class RpcReadIntegrationTests
 {
+    private static readonly TokenBucketRateLimiter RequestLimiter = new(new TokenBucketRateLimiterOptions
+    {
+        TokenLimit = 1,
+        QueueLimit = 128,
+        QueueProcessingOrder = QueueProcessingOrder.OldestFirst,
+        ReplenishmentPeriod = TimeSpan.FromMilliseconds(500),
+        TokensPerPeriod = 1,
+        AutoReplenishment = true,
+    });
+
     // USDC: a long-lived, heavily used SPL mint with stable, assertable properties (6 decimals).
     private static readonly PublicKey UsdcMint = PublicKey.Parse("EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v");
     private static readonly PublicKey TokenProgram = PublicKey.Parse("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA");
@@ -25,7 +36,10 @@ public static class RpcReadIntegrationTests
     private static ServiceProvider CreateProvider()
     {
         var services = new ServiceCollection();
-        services.AddSolanaRpc(IntegrationEnvironment.HttpEndpoint);
+        services.AddSolanaRpc(
+            options => options.Endpoint = IntegrationEnvironment.HttpEndpoint,
+            resilience => resilience.RateLimiter.RateLimiter =
+                arguments => RequestLimiter.AcquireAsync(1, arguments.Context.CancellationToken));
         return services.BuildServiceProvider();
     }
 

tests/SolSharp.IntegrationTests/WsIntegrationTests.cs

@@ -1,3 +1,4 @@
+using System.Diagnostics;
 using FluentAssertions;
 using NUnit.Framework;
 using SolSharp.Core.Constants;
@@ -14,13 +15,19 @@ namespace SolSharp.IntegrationTests;
 /// </summary>
 public static class WsIntegrationTests
 {
+    private static readonly TimeSpan MinimumProbeStartInterval = TimeSpan.FromMilliseconds(500);
+    private static readonly SemaphoreSlim ProbeGate = new(1, 1);
+    private static readonly Stopwatch ProbeClock = Stopwatch.StartNew();
+    private static TimeSpan _nextProbeStart;
+
     // Subjects picked for constant on-chain churn, so a healthy node delivers the first notification within
     // seconds: the SPL Token program sees near-continuous traffic, and the Clock sysvar changes every slot.
     private static readonly PublicKey TokenProgram = PublicKey.Parse(SolanaProgramIds.TokenProgram);
     private static readonly PublicKey Clock = PublicKey.Parse(Sysvars.Clock);
 
     [TestFixture]
     [Category("Integration")]
+    [NonParallelizable]
     public sealed class SubscribeSlots
     {
         [Test]
@@ -36,6 +43,7 @@ public Task ReceivesNotification() => ProbeAsync(async (client, token) =>
 
     [TestFixture]
     [Category("Integration")]
+    [NonParallelizable]
     public sealed class SubscribeRoots
     {
         [Test]
@@ -51,6 +59,7 @@ public Task ReceivesRootedSlot() => ProbeAsync(async (client, token) =>
 
     [TestFixture]
     [Category("Integration")]
+    [NonParallelizable]
     public sealed class SubscribeLogs
     {
         [Test]
@@ -65,6 +74,7 @@ public Task ReceivesLogsMentioningTheTokenProgram() => ProbeAsync(async (client,
 
     [TestFixture]
     [Category("Integration")]
+    [NonParallelizable]
     public sealed class SubscribeAccount
     {
         [Test]
@@ -80,6 +90,7 @@ public Task ReceivesAClockUpdate() => ProbeAsync(async (client, token) =>
 
     [TestFixture]
     [Category("Integration")]
+    [NonParallelizable]
     public sealed class SubscribeParsedAccount
     {
         [Test]
@@ -94,22 +105,37 @@ public Task DecodesAClockUpdate() => ProbeAsync(async (client, token) =>
         });
     }
 
-    // Connects a fresh client, runs the probe under a 30s deadline, and applies the shared integration-mode
-    // policy. Ordinary runs turn transport flakiness into an inconclusive result; strict release runs fail.
-    // A real assertion failure is never classified as transient, so it always fails the test.
+    // Serializes live probes and spaces their starts so independently scheduled fixtures cannot burst a
+    // provider's WebSocket request limit. The gate covers the complete subscription lifetime, including
+    // unsubscribe, while each probe's 30s deadline starts only after it owns the gate.
     private static async Task ProbeAsync(Func<SolanaWsClient, CancellationToken, Task> probe)
     {
+        await ProbeGate.WaitAsync();
+
         try
         {
-            await using var client = new SolanaWsClient();
-            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            var delay = _nextProbeStart - ProbeClock.Elapsed;
+            if (delay > TimeSpan.Zero)
+                await Task.Delay(delay);
+
+            _nextProbeStart = ProbeClock.Elapsed + MinimumProbeStartInterval;
+
+            try
+            {
+                await using var client = new SolanaWsClient();
+                using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
 
-            await client.ConnectAsync(new Uri(IntegrationEnvironment.WsEndpoint), timeout.Token);
-            await probe(client, timeout.Token);
+                await client.ConnectAsync(new Uri(IntegrationEnvironment.WsEndpoint), timeout.Token);
+                await probe(client, timeout.Token);
+            }
+            catch (Exception exception)
+            {
+                IntegrationEnvironment.RethrowOrInconclusive(exception);
+            }
         }
-        catch (Exception exception)
+        finally
         {
-            IntegrationEnvironment.RethrowOrInconclusive(exception);
+            ProbeGate.Release();
         }
     }
 }

tests/SolSharp.Rpc.Tests/Streaming/SolanaWsClientTests.cs

@@ -6,6 +6,7 @@
 using NUnit.Framework;
 using SolSharp.Core.Constants;
 using SolSharp.Core.Primitives;
+using SolSharp.Rpc.Protocol;
 using SolSharp.Rpc.Streaming;
 
 namespace SolSharp.Rpc.Tests.Streaming;
@@ -531,7 +532,7 @@ await WaitUntil(() => fake.SentSnapshot().Any(message =>
     public sealed class SubscribeRejection
     {
         [Test]
-        public async Task ErrorResponse_FaultsTheSubscribeCall()
+        public async Task ErrorResponse_PreservesRpcErrorAsInnerException()
         {
             // Arrange
             var fake = new FakeWebSocketConnection();
@@ -545,12 +546,16 @@ public async Task ErrorResponse_FaultsTheSubscribeCall()
 
             await WaitUntil(() => fake.Sent.Count > 0);
             fake.PushFromServer(
-                """{"jsonrpc":"2.0","error":{"code":-32602,"message":"Too many subscriptions"},"id":1}""");
+                """{"jsonrpc":"2.0","error":{"code":-32602,"message":"Too many subscriptions","data":{"limit":15}},"id":1}""");
 
             // Assert
             var act = async () => await subscribe;
-            (await act.Should().ThrowAsync<InvalidOperationException>())
-                .Which.Message.Should().Contain("-32602").And.Contain("Too many subscriptions");
+            var exception = (await act.Should().ThrowAsync<InvalidOperationException>()).Which;
+            exception.Message.Should().Contain("-32602").And.Contain("Too many subscriptions");
+            var rpcException = exception.InnerException.Should().BeOfType<RpcException>().Subject;
+            rpcException.Code.Should().Be(-32602);
+            rpcException.ErrorData.Should().NotBeNull();
+            rpcException.ErrorData!.Value.GetProperty("limit").GetInt32().Should().Be(15);
         }
 
         [Test]
@@ -574,7 +579,8 @@ public async Task ErrorResponse_DoesNotDisturbOtherSubscriptions()
             fake.PushFromServer("""{"jsonrpc":"2.0","error":{"code":-32000,"message":"nope"},"id":2}""");
 
             var act = async () => await second;
-            await act.Should().ThrowAsync<InvalidOperationException>();
+            var exception = (await act.Should().ThrowAsync<InvalidOperationException>()).Which;
+            exception.InnerException.Should().BeOfType<RpcException>();
 
             // Assert: the first subscription still delivers.
             fake.PushFromServer(
@@ -586,6 +592,7 @@ public async Task ErrorResponse_DoesNotDisturbOtherSubscriptions()
 
         [TestCase("{\"jsonrpc\":\"2.0\",\"error\":{},\"id\":1}")]
         [TestCase("{\"jsonrpc\":\"2.0\",\"error\":\"nope\",\"id\":1}")]
+        [TestCase("{\"jsonrpc\":\"2.0\",\"error\":{\"code\":2147483648,\"message\":\"nope\"},\"id\":1}")]
         [TestCase("{\"jsonrpc\":\"2.0\",\"result\":7,\"error\":{\"code\":-1,\"message\":\"nope\"},\"id\":1}")]
         [TestCase("{\"jsonrpc\":\"2.0\",\"id\":1}")]
         [TestCase("{\"jsonrpc\":\"2.0\",\"result\":7,\"id\":\"1\"}")]