StreamingMessageHandler concurrent sends crash NetMQ with "Cannot close an uninitialised Msg"

Expected Behavior

StreamingMessageHandler should safely accept packets from LEAN's algorithm and result threads. With a connected receiver, concurrent calls should deliver intact packets without corrupting the NetMQ socket or terminating the process.

Actual Behavior

Concurrent calls to Send() / Transmit() share one PushSocket without synchronization. We observed an intermittent process crash during an interactive backtest, shortly after algorithm completion. A standalone test using the real handler reproduces the same exception without an algorithm, Python, market data, a brokerage connection or a web browser:

Unhandled exception. NetMQ.FaultException: Cannot close an uninitialised Msg.
   at NetMQ.Msg.Close()
   at NetMQ.Core.Transports.EncoderBase.Encode(ByteArraySegment& data, Int32 size)
   at NetMQ.Core.Transports.StreamEngine.BeginSending()
   at NetMQ.Core.Transports.StreamEngine.Handle(Action action, SocketError socketError, Int32 bytesTransferred)
   at NetMQ.Core.Transports.StreamEngine.FeedAction(Action action, SocketError socketError, Int32 bytesTransferred)
   at NetMQ.Core.Transports.StreamEngine.ActivateOut()
   at NetMQ.Core.SessionBase.ReadActivated(Pipe pipe)
   at NetMQ.Core.Pipe.ProcessActivateRead()
   at NetMQ.Core.ZObject.ProcessCommand(Command cmd)
   at NetMQ.Core.IOThread.Ready()
   at NetMQ.Core.IOThreadMailbox.RaiseEvent()
   at NetMQ.Core.Utils.Proactor.Loop()
   at System.Threading.Thread.StartCallback()

The exception occurs on NetMQ's I/O thread and terminates the process. Catching exceptions around an individual caller's Send() does not address the concurrent socket access. The viewer can lose the final stream even when algorithm execution has finished.

Potential Solution

Serialize access to the socket in StreamingMessageHandler, where all packet send paths meet. Also dispose the socket under the same lock; the existing Dispose() is empty, although LeanEngineSystemHandlers.Dispose() already calls Notify.DisposeSafely().

The relevant paths at the tested LEAN revision are:

• BacktestingResultHandler.Run(): the result thread drains queued packets and calls MessagingHandler.Send(packet) at line 133.
• AlgorithmManager.Run(): algorithm lifecycle/status paths call results.SendStatusUpdate(...), including completion.
• BacktestingResultHandler.SendStatusUpdate(): directly calls MessagingHandler.Send(statusPacket) rather than going through that queue.
• StreamingMessageHandler.Transmit() on checked master: all these calls reach _server.SendMultipartMessage(message) without mutual exclusion.

This patch passed our before/after test:

--- a/Messaging/StreamingMessageHandler.cs
+++ b/Messaging/StreamingMessageHandler.cs
@@ -35,6 +35,7 @@
     {
         private string _port;
         private PushSocket _server;
+        private readonly object _socketLock = new object();
         private AlgorithmNodePacket _job;
         private OrderEventJsonConverter _orderEventJsonConverter;
 
@@ -100,7 +101,13 @@
 
             message.Append(payload);
 
-            _server.SendMultipartMessage(message);
+            // NetMQ sockets require exclusive access, including a memory barrier
+            // when ownership moves between the algorithm and result threads.
+            lock (_socketLock)
+            {
+                if (_server == null) throw new ObjectDisposedException(nameof(StreamingMessageHandler));
+                _server.SendMultipartMessage(message);
+            }
         }
 
         /// <summary>
@@ -126,6 +133,11 @@
         /// </summary>
         public void Dispose()
         {
+            lock (_socketLock)
+            {
+                _server?.Dispose();
+                _server = null;
+            }
         }
     }
 }

The lock provides exclusive socket access and the memory barrier when ownership moves between threads. JSON serialization remains outside the lock. This keeps the existing synchronous send/backpressure behavior and wire format; it does not add a new queue, change NetMQ versions or change algorithm behavior. The disposal guard makes repeated disposal safe and rejects sends after disposal.

A dedicated socket-owning thread/poller is another possible design, but would require queue, shutdown/drain and error-propagation decisions. The patch above is the smaller fix we tested. We have not tested disconnected-peer/high-water-mark behavior or live brokerage runs; existing blocking sends with no receiver are not resolved by this patch.

Reproducing the Problem

The test below sends 16,000 uniquely numbered DebugPackets through the real handler using up to eight concurrent producers. A separate thread owns a real TCP PullSocket and checks for duplicate/missing messages. Port 5555 must be free. Run the test in a separate process because the unpatched failure is an unhandled background-thread exception. Timing is nondeterministic; repeat the process if a run happens to pass.

With .NET SDK 10 installed, build a clean checkout at the master commit checked on September 13, 2026:

git clone https://github.com/QuantConnect/Lean.git Lean-stream-repro
cd Lean-stream-repro
git checkout --detach 6eb389012d73c364547d61546ff822fc8432dee2
dotnet build Launcher/QuantConnect.Lean.Launcher.csproj -c Release --verbosity quiet /p:RunAnalyzers=false -m:2

Save the following two files in a separate directory outside the LEAN checkout, for example /tmp/lean-stream-repro/.

Stress.csproj:

<Project Sdk="Microsoft.NET.Sdk">
  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <TargetFramework>net10.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
  </PropertyGroup>
  <ItemGroup>
    <Reference Include="$(LeanOutput)/*.dll" />
    <Compile Include="$(HandlerSource)" />
  </ItemGroup>
</Project>

Program.cs:

using NetMQ;
using NetMQ.Sockets;
using Newtonsoft.Json.Linq;
using QuantConnect.Configuration;
using QuantConnect.Messaging;
using QuantConnect.Packets;

const int count = 16000;
Config.Set("desktop-http-port", "5555");
using var handler = new StreamingMessageHandler();
handler.Initialize(null);
var receiver = Task.Factory.StartNew(() =>
{
    using var socket = new PullSocket(">tcp://127.0.0.1:5555");
    var seen = new HashSet<int>();
    while (seen.Count < count)
    {
        if (!socket.TryReceiveFrameString(TimeSpan.FromSeconds(30), out var frame))
            throw new Exception($"Only received {seen.Count}/{count} packets");
        var packet = JObject.Parse(frame);
        if (packet["Message"] != null && !seen.Add(int.Parse((string)packet["Message"])))
            throw new Exception("Duplicate packet");
    }
}, TaskCreationOptions.LongRunning);
handler.SetAuthentication(new BacktestNodePacket());
Parallel.For(0, count, new ParallelOptions { MaxDegreeOfParallelism = 8 }, i =>
    handler.Send(new DebugPacket(1, "stress", "", i.ToString())));
receiver.GetAwaiter().GetResult();
handler.Dispose();
Console.WriteLine($"STREAM_STRESS_PASS {count}");

From the LEAN checkout, run:

timeout 90s dotnet run --project /tmp/lean-stream-repro/Stress.csproj -c Release \
  -p:LeanOutput="$PWD/Launcher/bin/Release" \
  -p:HandlerSource="$PWD/Messaging/StreamingMessageHandler.cs"

timeout is the GNU/Linux utility and bounds the whole process, including a blocked producer. The expected CS0436 warning says the directly compiled StreamingMessageHandler is preferred over the type in QuantConnect.Messaging.dll. That is intentional: it lets this test change only the handler source while keeping all other compiled dependencies fixed.

Apply the patch above to Messaging/StreamingMessageHandler.cs and run the same command again. A passing run prints:

STREAM_STRESS_PASS 16000

No LEAN engine rebuild is needed between those two runs because HandlerSource compiles the handler directly into the test executable. Repeat the patched run to exercise different schedules.

How we found and isolated it

1. While evaluating smaller Docker runtimes for local interactive backtesting, one fast run crashed after algorithm completion with the NetMQ stack above. Two immediate repetitions and an original-image control passed, so the initial observation alone did not establish a runtime regression or a reliable failure rate.
2. We traced the send callers and found that status sends from the algorithm path can overlap the result thread's queued sends. Both use the same unsynchronized PushSocket.
3. We reduced the workload to the standalone C# handler test above. The original handler crashed immediately with the same stack, with no Python, data, strategy or viewer involved.
4. We checked current master 6eb389012d73c364547d61546ff822fc8432dee2. Its handler source is identical to the unpatched handler at our pinned revision after newline normalization, and its messaging project still references NetMQ 4.0.1.6. Compiling the current-master handler into the isolated test reproduced the same crash.
5. Using the same test image and dependency assemblies, changing only that handler source to the patched version passed three consecutive runs of 16,000 messages. Our additional lifecycle check also verified repeated Dispose() and rejection of a send after disposal.

Broader integration checks with the patched image passed. The completed small option fixture delivered the same packet-type counts before and after the handler fix. These checks supplement the isolated reproducer; they are not a claim of exhaustive transport testing.

System Information

• Host: Windows with Docker Desktop; test processes ran in Linux x86-64 containers.
• Isolated before/after test runtime: Ubuntu Noble, .NET SDK 10.0.401 / .NET runtime 10.0.12, two CPUs and 1 GiB container memory.
• NetMQ: 4.0.1.6, unchanged before/after. QuantConnect.Messaging.csproj on checked master declares the same version.
• Original integrated LEAN revision: 8ee075a39918f2df6fe9e0a5944e366fb60d10dc, LEAN 2.5.0.0.
• Current-master handler checked and reproduced: 6eb389012d73c364547d61546ff822fc8432dee2 on September 13, 2026. We tested that handler in isolation against the pinned LEAN dependency assemblies; we did not run the entire current-master engine or its full test suite.
• Both original Debug assemblies and a directly compiled Release handler exhibited the failure. No Python runtime is initialized by the reproducer.

Checklist

• I have completely filled out this template
• I have confirmed that this issue exists on the current master branch (handler source at the commit above, isolated reproduction; full engine scope noted above)
• I have confirmed that this is not a duplicate issue by searching issues (no open matches; the three closed matches concern other topics)
• I have provided detailed steps to reproduce the issue

Start with Messaging/StreamingMessageHandler.cs, then trace the send paths from BacktestingResultHandler.Run(), AlgorithmManager.Run(), and BacktestingResultHandler.SendStatusUpdate(). Run the standalone stress reproduction against the current handler, checking for the NetMQ crash and packet loss. Done means concurrent sends complete with STREAM_STRESS_PASS 16000 and repeated disposal and post-disposal behavior match the stated expectations.