bug: Stream(async)Hubs are not thread-safe and produce ArgumentOutOfBoundsExceptions in the real world

[JGronholz]

• Streaming hubs are designed for async operation (websockets/REST)

You can reproduce this by having copilot use any hub in the real-world.
More hubs == more errors.
Each new error == same bounds-checking applied to another hub
Eventually, you'll end up at my branch.

The root-problem is that they aren't being tested in a typical use-case (websocket feed).

[JGronholz]

@DaveSkender If you change your mind, and want a peek at my code, I'm still willing to PR my branch for your review, but copilot will fix this easily, once it realizes it's an issue. I just know you won't be happy of the perf hit...

[DaveSkender]

Actually, I think this is a valid concern worth keeping in the backlog.
The problem I have is not being able to reproduce the issue you're suggesting, so if you have a use case -- even if some pseudocode explanation -- that'd help me visualize the problem better.

With that said, I may be able to do something with the SSE rig I have started in:

• feat: Add StrategyGroup with back-pair alignment #1899

Though, don't think that'd reproduce it either because WebSocket event handling (e.g. StreamReader) isn't innately async, you get one event at a time per subscription ... at least, that's my hypothesis. You're right, more testing needed.

I think this would only be a problem if you subscribed to the same WebSocket several times and tried to push them in parallel to a single hub, which wouldn't make much sense from a design perspective.

[JGronholz]

I think this would only be a problem if you subscribed to the same WebSocket several times and tried to push them in parallel to a single hub, which wouldn't make much sense from a design perspective.

This was the really confusing part, because I disabled chunks of my app in series trying to find the root.
I ended up with only one symbol/interval subscribed, feeding only one QuoteHub.

• It should work
• Copilot agrees
• It reliably fails after 5-60 seconds (nothing resembling deterministic)

The solution:

• If you experience concurrency issues, you're having concurrency issues and should synchronize access to the hubs.

(Reminder to RTFM since the library does not claim to be thread-safe.)

[JGronholz]

Using the Coinbase.Net package from JKorf:

<PackageReference Include="JKorf.Coinbase.Net" Version="2.11.3" /> //Latest stable is 3.2.2

Depends on CryptoExchange.Net (by JKorf).
Using version 9.13.0.
Latest stable is 10.2.5

I might be fighting known issues with deprecated code.

[DaveSkender]

Which endpoints are you ingesting, what rate, etc?
I'll try to repro with that library.

[JGronholz]

All defaults, except the api key.
I believe the klines feed does not require authentication.

await _socketClient.AdvancedTradeApi.SubscribeToKlineUpdatesAsync(symbol, evt => HandleKlines(Symbol.Symbol, evt));

https://docs.cdp.coinbase.com/coinbase-app/advanced-trade-apis/websocket/websocket-channels#candles-channel

The lib is async, but not concurrent.
You can only subscribe to the 5m klines.
They come in every few seconds. (~1 per 5 seconds)
I don't really understand...
Copilot suspects the lib uses non-deterministic methods under the hood, but didn't verify, and nether did I.

[DaveSkender]

reminder to RTFM since the library does not claim to be thread-safe

And you're right about that. You'd be surprised how much of native .NET code is not thread safe.

Here's a draft of something I'm considering for the documentation site (a47ad26)

---

Thread safety and concurrency

Understanding thread-safety is critical when working with streaming indicators and live data feeds. This section clarifies the library's threading model and provides guidance on safely integrating with real-time data sources.

see more

Library threading model

Stock.Indicators processes stream events serially - one event at a time. The library is designed for single-threaded sequential processing and is not thread-safe by default.

• Series style: Results are immutable and thread-safe; calculations themselves are not thread-safe
• Buffer lists: Not thread-safe; synchronize external access if sharing across threads
• Stream hubs: Not thread-safe; designed for single-threaded inputs like WebSocket/SSE

A StreamHub, QuoteHub, or BufferList instance reads one quote at a time and is not meant to be shared across threads without external synchronization.

Why streaming is often single-threaded

Thread-safety limitations in streaming contexts are normal and expected throughout the .NET ecosystem and common data feed technologies:

Common streaming technologies:

• WebSocket connections: Typically maintain a single connection with sequential message delivery. The underlying TCP stream is inherently sequential.
• Server-Sent Events (SSE): HTTP-based event streams deliver events one at a time over a single connection.
• Stock quote feeds (IEX, Alpaca, Interactive Brokers, etc.): Most real-time market data APIs deliver quotes sequentially over WebSocket or similar connections.
• SignalR hubs: .NET's real-time framework delivers messages to hub methods serially by default.
• gRPC streams: Bidirectional streaming maintains message order within a single stream.

Why this design is optimal:

• Order preservation: Financial calculations require chronological processing; parallel processing would require complex synchronization to maintain order
• Performance: Serial processing eliminates lock contention and coordination overhead for the common case
• Simplicity: Most streaming data sources are inherently sequential; the library matches this natural flow

Idiomatic usage patterns (no locks needed)

WebSocket example with serial processing

using System.Net.WebSockets;
using Skender.Stock.Indicators;

QuoteHub<Quote> quoteHub = new();
SmaHub<Quote> smaHub = quoteHub.ToSma(20);

async Task ProcessWebSocketStream(ClientWebSocket ws, string uri)
{
    await ws.ConnectAsync(new Uri(uri), CancellationToken.None);
    
    byte[] buffer = new byte[4096];
    
    while (ws.State == WebSocketState.Open)
    {
        var result = await ws.ReceiveAsync(
            new ArraySegment<byte>(buffer), 
            CancellationToken.None);
        
        if (result.MessageType == WebSocketMessageType.Text)
        {
            // parse quote from JSON message (your method)
            Quote quote = ParseQuoteFromJson(buffer, result.Count);
            
            // process serially - no locks needed
            quoteHub.Add(quote);
            
            var latest = smaHub.Results.LastOrDefault();
            // use results
        }
    }
}

Server-Sent Events (SSE) example

using Skender.Stock.Indicators;

QuoteHub<Quote> quoteHub = new();
RsiHub<Quote> rsiHub = quoteHub.ToRsi(14);

async Task ProcessSSEStream(string sseEndpoint)
{
    using var client = new HttpClient();
    using var stream = await client.GetStreamAsync(sseEndpoint);
    using var reader = new StreamReader(stream);
    
    while (!reader.EndOfStream)
    {
        string? line = await reader.ReadLineAsync();
        
        if (line?.StartsWith("data:") == true)
        {
            // parse quote from SSE data line (your method)
            Quote quote = ParseQuoteFromSSE(line);
            
            // process serially
            quoteHub.Add(quote);
            
            var latest = rsiHub.Results.LastOrDefault();
            // use results
        }
    }
}

Multiple independent streams

If you have multiple indicators that don't need synchronization, create separate instances per thread:

// thread 1: processes MSFT
Task.Run(() => {
    QuoteHub<Quote> msftHub = new();
    SmaHub<Quote> msftSma = msftHub.ToSma(20);
    
    foreach (var quote in msftStream)
    {
        msftHub.Add(quote);  // no locks needed
    }
});

// thread 2: processes AAPL
Task.Run(() => {
    QuoteHub<Quote> aaplHub = new();
    SmaHub<Quote> aaplSma = aaplHub.ToSma(20);
    
    foreach (var quote in aaplStream)
    {
        aaplHub.Add(quote);  // no locks needed
    }
});

Troubleshooting concurrent access

If your application architecture requires sharing a hub instance across threads or processing quotes from multiple concurrent sources, you'll need to add external synchronization. This section shows how to diagnose and fix concurrency issues.

Common symptoms:

• Inconsistent or corrupted results
• InvalidOperationException from collection modification
• Race conditions or unexpected behavior
• Debugging shows multiple threads accessing the same hub

Using locks for shared access

If you must share a hub across threads, use a dedicated lock object:

using Skender.Stock.Indicators;

QuoteHub<Quote> quoteHub = new();
SmaHub<Quote> smaHub = quoteHub.ToSma(20);

// dedicated lock object (never lock on quoteHub directly)
private readonly object _hubLock = new();

// thread 1: processing quotes
void ProcessQuotes(Quote quote)
{
    lock (_hubLock)
    {
        quoteHub.Add(quote);
    }
}

// thread 2: reading results
void ReadResults()
{
    lock (_hubLock)
    {
        var latest = smaHub.Results.LastOrDefault();
        // use latest result
    }
}

[!WARNING]
Never lock on the hub instance directly

Do NOT use lock (quoteHub) - always create a dedicated private readonly object for locking. Locking on public instances can cause deadlocks and violates encapsulation.

Using Channel for producer-consumer coordination

For high-throughput scenarios where you need to decouple quote reception from processing, use System.Threading.Channels:

using System.Threading.Channels;
using Skender.Stock.Indicators;

Channel<Quote> quoteChannel = Channel.CreateUnbounded<Quote>();

// producer thread: receives quotes from WebSocket
async Task ProduceQuotes(WebSocket ws)
{
    while (ws.State == WebSocketState.Open)
    {
        Quote quote = await ReceiveQuoteFromWebSocket(ws);
        await quoteChannel.Writer.WriteAsync(quote);
    }
    
    quoteChannel.Writer.Complete();
}

// consumer thread: processes quotes through indicators
async Task ConsumeQuotes()
{
    QuoteHub<Quote> quoteHub = new();
    SmaHub<Quote> smaHub = quoteHub.ToSma(20);
    
    await foreach (Quote quote in quoteChannel.Reader.ReadAllAsync())
    {
        quoteHub.Add(quote);  // serial processing
        
        var latest = smaHub.Results.LastOrDefault();
        // use results
    }
}

// start both tasks
await Task.WhenAll(
    ProduceQuotes(websocket),
    ConsumeQuotes()
);

When to use external synchronization

You need locks or channels when:

• Sharing a single hub instance across multiple threads
• Multiple threads call .Add(), .Insert(), or .Remove() on the same hub
• Reading results from one thread while another thread updates the hub
• Processing quotes from multiple concurrent sources into one hub

You DON'T need locks when:

• Processing a single sequential stream (WebSocket, SSE, file reading)
• Each thread has its own independent hub instances
• Using Series style for batch calculations (results are immutable)
• Reading from BufferList or StreamHub from the same thread that updates it

Performance considerations

• Locking overhead: For high-frequency updates (>10k quotes/sec), channel-based coordination typically outperforms simple locks
• Lock contention: Minimize lock hold time; read results outside the lock if possible
• False sharing: If running multiple independent hubs, ensure they don't share cache lines

---

How common is thread safety in .NET asynchronous environments?

Common real‑time technologies in .NET are built around asynchronous I/O, but most of them are not inherently thread‑safe. Each framework has its own rules for how concurrent access is allowed:

• SignalR: A SignalR server processes many requests concurrently. The Microsoft Q&A documentation notes that if you have shared state in your hub, you must make access to that state thread‑safe by locking around the shared resource. Splitting code across multiple hubs doesn’t change this, because the underlying threads can still access the same objects. Also, HubConnection objects on the client are not thread‑safe; instance members should not be called from multiple threads at once. For safe broadcasting, queue messages and use a single sending loop for each connection rather than firing SendAsync from several tasks concurrently.

• WebSockets: The underlying ClientWebSocket class allows only one send and one receive to be in progress at a time. The official API docs say that one send and one receive may run in parallel, but issuing multiple sends or multiple receives concurrently “is not supported and will result in undefined behaviour”. If you need to send messages from multiple producers, serialize calls to SendAsync (for example, via a ConcurrentQueue and a dedicated sender task).

• Server‑Sent Events (SSE): SSE streams are unidirectional and typically implemented by returning an IAsyncEnumerable<T> or reading from a StreamReader. A StreamReader is not thread‑safe by default. If multiple threads need to read from the same stream, wrap it using TextReader.Synchronized or provide each consumer with its own reader. In most SSE patterns, only one enumeration reads the stream, so events are delivered serially and no additional locking is needed. If you share a single event source among multiple clients, protect shared buffers with thread‑safe collections such as BlockingCollection<T>.

• Popular market‑data providers: Many third‑party libraries use WebSocket connections under the hood. Some, like JKorf’s Binance.Net/CryptoExchange.Net, document that only one subscriber should read a given WebSocket stream at a time, and their classes are not guaranteed to be thread‑safe. Use separate client instances per subscription or consult the library’s documentation for concurrency guidelines. When consuming data from these libraries, apply the same WebSocket rules above—queue outbound messages and avoid simultaneous sends or receives on the same socket.

Summary: Real‑time components in .NET are designed for asynchronous I/O but not for free‑form multithreaded access. Treat hubs, WebSocket clients and SSE stream readers as single‑consumer objects. Protect shared state with locks or concurrent collections, and queue messages so that only one send or receive call is active at a time. If you need to broadcast to multiple consumers, create separate connections or use thread‑safe collections to manage shared data.

Additional examples:

• docs: Update v3 documentation site #1905 (comment)

[DaveSkender]

I was able to reproduce! Thanks for the notes.

I’m testing a few options to internalize a few aspects of locking that should make this less annoying to handle by callers.

• refactor: Improve thread safety for live feeds #1927