fix: thread-safe RequestManager and surface stream-handler errors (v10.4.2)

CHANGES.md

@@ -1,5 +1,10 @@
 # Changes
 
+### 10.4.2.0 06/05/2026
+* Fix thread-unsafe request id assignment in `RequestManager` — concurrent requests on a single connection (e.g. `Task.WhenAll` over several `BookOffers`) could collide on the same id and throw `Response with id '$<guid>' is already pending` or drop a pending promise. Removed the shared `nextId` field; each call now generates its own `Guid` and registers via a single atomic `ConcurrentDictionary.TryAdd`, enabling parallel requests on one connection
+* Surface exceptions thrown by stream handlers (`OnLedgerClosed`, `OnTransaction`, etc.) through the `OnError` event instead of swallowing them into a debug trace — consumer bugs are now observable, while the message loop stays alive and a throwing `OnError` handler is contained
+* Clarify in XML docs that `Xrpl.Client.Exceptions.TimeoutException` is not `System.TimeoutException` (it derives from `XrplException`), to avoid mismatched `catch` clauses
+
 ### 10.4.1.0 05/28/2026
 * Fix `IouValue` (IOU token amount) parsing to accept a trailing decimal point (e.g. `"128700."`), aligning with `xrpl.js` / `ripple-binary-codec` and `rippled` `STAmount` reference behavior — previously the stricter validation regex rejected a value with no digits after the dot, breaking signing of transactions (e.g. `AMMDeposit` via WalletConnect) that carried such amounts
 * Relax IOU value regex fractional group from `(\.(\d+))?` to `(\.(\d*))?` while adding a `(?=\.?\d)` lookahead that still requires at least one mantissa digit — so trailing/leading dots (`"128700."`, `".5"`) parse but bare-dot inputs (`"."`, `".e10"`) are rejected, matching BigNumber; deduplicate the regex by reusing the single `IouValue.ValueRegex` constant in `AmountValue.cs` and `ExtenstionHelpers.cs`

Tests/Xrpl.Tests/Client/TestSubscribe.cs

@@ -257,6 +257,28 @@ public async Task TestEmitsBookChanges()
             Assert.AreEqual("LESS", change.AssetB.ToString());
         }
 
+        [TestMethod]
+        public async Task StreamHandlerException_IsSurfacedViaOnError()
+        {
+            TaskCompletionSource<string> errorReported = new TaskCompletionSource<string>();
+
+            runner.client.connection.OnLedgerClosed += _ =>
+                throw new InvalidOperationException("consumer handler bug");
+
+            runner.client.connection.OnError += (error, errorMessage, message, data) =>
+            {
+                errorReported.TrySetResult(errorMessage);
+                return Task.CompletedTask;
+            };
+
+            string jsonString = "{\"fee_base\":10,\"fee_ref\":10,\"ledger_hash\":\"B3980C722D71873D6708723E71B7A28C826BC66C58712ADCEC61603415305CD1\",\"ledger_index\":66093872,\"ledger_time\":683942720,\"reserve_base\":20000000,\"reserve_inc\":5000000,\"txn_count\":70,\"type\":\"ledgerClosed\",\"validated_ledgers\":\"65201743-66093872\"}";
+            await runner.client.connection.OnMessage(jsonString);
+
+            Task completed = await Task.WhenAny(errorReported.Task, Task.Delay(5000));
+            Assert.AreEqual(errorReported.Task, completed,
+                "Exception thrown by a stream handler was swallowed instead of surfaced via OnError");
+        }
+
         [TestMethod]
         public async Task TestEmitsServerStatus()
         {

Tests/Xrpl.Tests/Client/TestURequestManagerConcurrency.cs

@@ -0,0 +1,109 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+using Xrpl.Client;
+using Xrpl.Client.Exceptions;
+
+namespace Xrpl.Tests.ClientLib
+{
+    [TestClass]
+    public class TestURequestManagerConcurrency
+    {
+        private class FakeRequest
+        {
+            public Guid? Id { get; set; }
+        }
+
+        private static void RunConcurrent(Action body, out IReadOnlyList<Exception> errors)
+        {
+            int threadCount = Math.Max(8, Environment.ProcessorCount * 2);
+            const int iterationsPerThread = 500;
+
+            ConcurrentBag<Exception> collected = new ConcurrentBag<Exception>();
+            Barrier barrier = new Barrier(threadCount);
+            Thread[] threads = new Thread[threadCount];
+
+            for (int t = 0; t < threadCount; t++)
+            {
+                threads[t] = new Thread(() =>
+                {
+                    barrier.SignalAndWait();
+                    for (int i = 0; i < iterationsPerThread; i++)
+                    {
+                        try
+                        {
+                            body();
+                        }
+                        catch (Exception ex)
+                        {
+                            collected.Add(ex);
+                        }
+                    }
+                });
+            }
+
+            foreach (Thread thread in threads)
+                thread.Start();
+            foreach (Thread thread in threads)
+                thread.Join();
+
+            errors = collected.ToArray();
+        }
+
+        [TestMethod]
+        public void CreateRequest_ConcurrentCalls_AssignUniqueIdsWithoutCollision()
+        {
+            RequestManager manager = new RequestManager();
+            ConcurrentBag<Guid> ids = new ConcurrentBag<Guid>();
+
+            RunConcurrent(
+                () =>
+                {
+                    RequestManager.XrplRequest request = manager.CreateRequest(
+                        new Dictionary<string, object>(),
+                        System.Threading.Timeout.InfiniteTimeSpan);
+                    ids.Add(request.Id);
+                },
+                out IReadOnlyList<Exception> errors);
+
+            Assert.AreEqual(0, errors.Count,
+                $"Concurrent CreateRequest threw {errors.Count} exception(s): " +
+                string.Join(" | ", errors.Take(3).Select(e => e.Message)));
+
+            List<Guid> all = ids.ToList();
+            Assert.AreEqual(all.Count, all.Distinct().Count(),
+                "Concurrent CreateRequest assigned duplicate ids");
+        }
+
+        [TestMethod]
+        public void CreateGRequest_ConcurrentCalls_AssignUniqueIdsWithoutCollision()
+        {
+            RequestManager manager = new RequestManager();
+            ConcurrentBag<Guid> ids = new ConcurrentBag<Guid>();
+
+            RunConcurrent(
+                () =>
+                {
+                    RequestManager.XrplGRequest request = manager.CreateGRequest<object, FakeRequest>(
+                        new FakeRequest(),
+                        System.Threading.Timeout.InfiniteTimeSpan);
+                    ids.Add(request.Id);
+                },
+                out IReadOnlyList<Exception> errors);
+
+            Assert.AreEqual(0, errors.Count,
+                $"Concurrent CreateGRequest threw {errors.Count} exception(s): " +
+                string.Join(" | ", errors.Take(3).Select(e => e.Message)));
+
+            List<Guid> all = ids.ToList();
+            Assert.AreEqual(all.Count, all.Distinct().Count(),
+                "Concurrent CreateGRequest assigned duplicate ids");
+        }
+    }
+}

Xrpl/Client/Exceptions/XrplException.cs

@@ -96,7 +96,15 @@ public DisconnectedException(string message) : base(message)
     /// </summary>
     public class RippledNotInitializedException : XrplException { }
     /// <summary>
-    /// Exception thrown when xrpl.js times out.
+    /// Exception thrown when a request to rippled times out.
+    /// <para>
+    /// IMPORTANT: this is <c>Xrpl.Client.Exceptions.TimeoutException</c>, NOT
+    /// <see cref="System.TimeoutException"/>. It derives from <see cref="XrplException"/>, so a
+    /// <c>catch (System.TimeoutException)</c> will NOT catch it. Catch this type (or its base
+    /// <see cref="XrplException"/>) explicitly; if both <c>using System;</c> and
+    /// <c>using Xrpl.Client.Exceptions;</c> are in scope, fully-qualify the type to avoid catching
+    /// the wrong one.
+    /// </para>
     /// </summary>
     public class TimeoutException : XrplException
     {

Xrpl/Client/RequestManager.cs

@@ -42,7 +42,6 @@ public class XrplGRequest
             public Task<object> Promise { get; set; }
         }
 
-        private Guid nextId = Guid.NewGuid();
         private readonly ConcurrentDictionary<Guid, Timer> timeoutsAwaitingResponse = new ConcurrentDictionary<Guid, Timer>();
         private readonly ConcurrentDictionary<Guid, TaskInfo> promisesAwaitingResponse = new ConcurrentDictionary<Guid, TaskInfo>();
         private readonly JsonSerializerOptions serializerOptions = XrplJsonOptions.Default;
@@ -166,35 +165,25 @@ public XrplGRequest CreateGRequest<T, R>(R request, TimeSpan timeout, Cancellati
                     $"Timeout must be positive or Timeout.InfiniteTimeSpan, but was {timeout.TotalSeconds:F1}s");
             }
 
-            Guid newId;
             var info = request.GetType().GetProperty("Id");
-            if (info.GetValue(request) == null)
-            {
-                newId = this.nextId;
-                this.nextId = Guid.NewGuid();
-            }
-            else
-            {
-                newId = (Guid)info.GetValue(request);
-            }
+            object existingId = info.GetValue(request);
+            Guid newId = existingId == null ? Guid.NewGuid() : (Guid)existingId;
 
             info.SetValue(request, newId, null);
 
             string newRequest = JsonSerializer.Serialize(request, serializerOptions);
 
-            if (this.promisesAwaitingResponse.ContainsKey(newId))
-            {
-                throw new XrplException($"Response with id '${newId}' is already pending");
-            }
-
             TaskCompletionSource<object> task = new TaskCompletionSource<object>();
             TaskInfo taskInfo = new TaskInfo();
             taskInfo.TaskId = newId;
             taskInfo.TaskCompletionResult = task;
             taskInfo.RemoveUponCompletion = true;
             taskInfo.Type = typeof(T);
 
-            promisesAwaitingResponse.TryAdd(newId, taskInfo);
+            if (!promisesAwaitingResponse.TryAdd(newId, taskInfo))
+            {
+                throw new XrplException($"Response with id '${newId}' is already pending");
+            }
 
             if (cancellationToken.CanBeCanceled)
             {
@@ -249,35 +238,24 @@ public XrplRequest CreateRequest(Dictionary<string, object> request, TimeSpan ti
                     $"Timeout must be positive or Timeout.InfiniteTimeSpan, but was {timeout.TotalSeconds:F1}s");
             }
 
-            Guid newId;
             var hasId = request.TryGetValue("id", out var id);
-            if (!hasId)
-            {
-                newId = this.nextId;
-                this.nextId = Guid.NewGuid();
-            }
-            else
-            {
-                newId = (Guid)id;
-            }
+            Guid newId = hasId ? (Guid)id : Guid.NewGuid();
 
             request["id"] = newId;
 
             string newRequest = JsonSerializer.Serialize(request, serializerOptions);
 
-            if (this.promisesAwaitingResponse.ContainsKey(newId))
-            {
-                throw new XrplException($"Response with id '${newId}' is already pending");
-            }
-
             TaskCompletionSource<Dictionary<string, object>> task = new TaskCompletionSource<Dictionary<string, object>>();
             TaskInfo taskInfo = new TaskInfo();
             taskInfo.TaskId = newId;
             taskInfo.TaskCompletionResult = task;
             taskInfo.RemoveUponCompletion = true;
             taskInfo.Type = typeof(Dictionary<string, object>);
 
-            promisesAwaitingResponse.TryAdd(newId, taskInfo);
+            if (!promisesAwaitingResponse.TryAdd(newId, taskInfo))
+            {
+                throw new XrplException($"Response with id '${newId}' is already pending");
+            }
 
             if (cancellationToken.CanBeCanceled)
             {

Xrpl/Client/connection.cs

@@ -2419,14 +2419,14 @@ private void StartMessageProcessor()
                         {
                             if (cts.Token.IsCancellationRequested)
                                 return;
-                            
+
                             try
                             {
                                 await ProcessStreamMessageAsync(message).ConfigureAwait(false);
                             }
                             catch (Exception ex)
                             {
-                                Debug.WriteLine($"{DateTime.Now}Stream message processing error: {ex.Message}");
+                                await NotifyStreamProcessingErrorAsync(ex, message).ConfigureAwait(false);
                             }
                         }
                     }
@@ -2755,7 +2755,39 @@ private async Task ProcessStreamMessageFireAndForgetAsync(string message)
         }
         catch (Exception ex)
         {
-            Debug.WriteLine($"{DateTime.Now}Stream message processing error: {ex.Message}");
+            await NotifyStreamProcessingErrorAsync(ex, message).ConfigureAwait(false);
+        }
+    }
+
+    /// <summary>
+    /// Surfaces an exception raised while processing a stream message — including exceptions
+    /// thrown by consumer stream handlers (e.g. <see cref="OnLedgerClosed"/>, <see cref="OnTransaction"/>) —
+    /// through the <see cref="OnError"/> event instead of swallowing it into a debug trace, so consumer
+    /// bugs are observable. The message loop is always kept alive: cancellation is ignored, and an
+    /// exception thrown by the <see cref="OnError"/> handler itself is contained.
+    /// </summary>
+    private async Task NotifyStreamProcessingErrorAsync(Exception ex, string message)
+    {
+        Debug.WriteLine($"{DateTime.Now}Stream message processing error: {ex.Message}");
+
+        if (ex is OperationCanceledException)
+        {
+            return;
+        }
+
+        var handler = OnError;
+        if (handler is null)
+        {
+            return;
+        }
+
+        try
+        {
+            await handler.Invoke(error: "error", errorMessage: "streamHandlerError", message: ex.Message, data: message).ConfigureAwait(false);
+        }
+        catch (Exception notifyEx)
+        {
+            Debug.WriteLine($"{DateTime.Now}OnError handler threw while reporting stream processing error: {notifyEx.Message}");
         }
     }
 }

Xrpl/Xrpl.csproj

@@ -14,7 +14,7 @@
     <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
     <PackageProjectUrl>https://github.com/StaticBit-io/XrplCSharp</PackageProjectUrl>
     <Title>XrplCSharp</Title>
-    <PackageVersion>10.4.1.0</PackageVersion>
+    <PackageVersion>10.4.2.0</PackageVersion>
   </PropertyGroup>
 
   <PropertyGroup>