From 0f0c204baf6ce061f3a59ef5d56b37ab4650f1b8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 21 May 2026 11:01:45 +0000 Subject: [PATCH 1/6] Initial plan From 493d63a8f64add71fd21fd411a1b9e432ed67785 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 21 May 2026 11:22:22 +0000 Subject: [PATCH 2/6] Fix null peer identity handling in StreamEngine handshake Agent-Logs-Url: https://github.com/zeromq/netmq/sessions/f5cde7c6-6d90-4da1-9926-4e0396ec2b03 Co-authored-by: drewnoakes <350947+drewnoakes@users.noreply.github.com> --- src/NetMQ.Tests/MechanismTests.cs | 51 ++++++++++++++++++++++- src/NetMQ/Core/Transports/StreamEngine.cs | 12 ++++-- 2 files changed, 59 insertions(+), 4 deletions(-) diff --git a/src/NetMQ.Tests/MechanismTests.cs b/src/NetMQ.Tests/MechanismTests.cs index d9469d18..0883b197 100644 --- a/src/NetMQ.Tests/MechanismTests.cs +++ b/src/NetMQ.Tests/MechanismTests.cs @@ -1,5 +1,9 @@ -using System.Text; +using System.Reflection; +using System.Runtime.Serialization; +using System.Text; +using NetMQ.Core; using NetMQ.Core.Mechanisms; +using NetMQ.Core.Transports; using Xunit; namespace NetMQ.Tests @@ -43,6 +47,51 @@ public void IsCommandShouldReturnFalseForInvalidCommand() Assert.False(mechanism.IsCommand("READY", ref msg)); } + [Fact] + public void MechanismReadyShouldHandleNullPeerIdentityWhenRecvIdentityIsEnabled() + { +#pragma warning disable SYSLIB0050 + var streamEngine = (StreamEngine)FormatterServices.GetUninitializedObject(typeof(StreamEngine)); +#pragma warning restore SYSLIB0050 + var options = new Options { RecvIdentity = true, HeartbeatInterval = 0 }; + var session = new RecordingSession(); + var mechanism = new NullMechanism(session, options) { PeerIdentity = null }; + + SetPrivateField(streamEngine, "m_options", options); + SetPrivateField(streamEngine, "m_session", session); + SetPrivateField(streamEngine, "m_mechanism", mechanism); + + var method = typeof(StreamEngine).GetMethod("MechanismReady", BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(method); + + var exception = Record.Exception(() => method!.Invoke(streamEngine, null)); + Assert.Null(exception); + Assert.Equal(0, session.LastPushedMessageSize); + } + + private static void SetPrivateField(object instance, string fieldName, object value) + { + var field = instance.GetType().GetField(fieldName, BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(field); + field!.SetValue(instance, value); + } + + private sealed class RecordingSession : SessionBase + { + public int LastPushedMessageSize { get; private set; } = -1; + + public RecordingSession() + : base(new IOThread(new Ctx(), 0), false, null!, new Options(), null!) + { + } + + public override PushMsgResult PushMsg(ref Msg msg) + { + LastPushedMessageSize = msg.Size; + return PushMsgResult.Ok; + } + } + // this test was used to validate the behavior prior to changing the validation logic in Mechanism.IsCommand // [Fact] // public void IsCommandShouldThrowWhenLengthByteExceedsSize() diff --git a/src/NetMQ/Core/Transports/StreamEngine.cs b/src/NetMQ/Core/Transports/StreamEngine.cs index 4089f817..08ec19b7 100644 --- a/src/NetMQ/Core/Transports/StreamEngine.cs +++ b/src/NetMQ/Core/Transports/StreamEngine.cs @@ -1199,8 +1199,14 @@ void MechanismReady () if (m_options.RecvIdentity) { Msg identity = new Msg(); - identity.InitPool(m_mechanism.PeerIdentity.Length); - identity.Put(m_mechanism.PeerIdentity, 0, m_mechanism.PeerIdentity.Length); + byte[] peerIdentity = m_mechanism.PeerIdentity; + if (peerIdentity == null) + identity.InitEmpty(); + else + { + identity.InitPool(peerIdentity.Length); + identity.Put(peerIdentity, 0, peerIdentity.Length); + } var pushResult = m_session.PushMsg(ref identity); if (pushResult == PushMsgResult.Full) { // If the write is failing at this stage with @@ -1384,4 +1390,4 @@ public void TimerEvent(int id) } } -} \ No newline at end of file +} From f841e0ccebb2111cf2140a2794c2e7bb2c5e9e1e Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 21 May 2026 11:54:31 +0000 Subject: [PATCH 3/6] Annotate StreamEngine peer identity path with nullable context Agent-Logs-Url: https://github.com/zeromq/netmq/sessions/bfe457f8-b95d-4df4-8cbc-0e45caa5cf65 Co-authored-by: drewnoakes <350947+drewnoakes@users.noreply.github.com> --- src/NetMQ/Core/Transports/StreamEngine.cs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/NetMQ/Core/Transports/StreamEngine.cs b/src/NetMQ/Core/Transports/StreamEngine.cs index 08ec19b7..31f2128d 100644 --- a/src/NetMQ/Core/Transports/StreamEngine.cs +++ b/src/NetMQ/Core/Transports/StreamEngine.cs @@ -1189,6 +1189,7 @@ PushMsgResult ProcessHandshakeCommand (ref Msg msg) return result; } +#nullable enable void MechanismReady () { if (m_options.HeartbeatInterval > 0) @@ -1199,8 +1200,8 @@ void MechanismReady () if (m_options.RecvIdentity) { Msg identity = new Msg(); - byte[] peerIdentity = m_mechanism.PeerIdentity; - if (peerIdentity == null) + byte[]? peerIdentity = m_mechanism.PeerIdentity; + if (peerIdentity is null) identity.InitEmpty(); else { @@ -1221,6 +1222,7 @@ void MechanismReady () m_nextMsg = PullAndEncode; m_processMsg = DecodeAndPush; } +#nullable disable PullMsgResult PullAndEncode (ref Msg msg) { From 1fff70762bcebcc3054d24c53f29cc1787e414f1 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 24 May 2026 03:14:23 +0000 Subject: [PATCH 4/6] test: make thread-safe cleanup receive loop non-blocking Agent-Logs-Url: https://github.com/zeromq/netmq/sessions/d30df06d-eaa6-46d7-b83a-7574408492b7 Co-authored-by: drewnoakes <350947+drewnoakes@users.noreply.github.com> --- src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs b/src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs index 12d889ce..36603777 100644 --- a/src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs +++ b/src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs @@ -56,7 +56,7 @@ public async Task ClosingServerWhileClientsConnectDoesNotCrash() msg.InitEmpty(); while (!cts.IsCancellationRequested) { - if (server.TryReceive(ref msg, TimeSpan.FromMilliseconds(1))) + if (server.TryReceive(ref msg, TimeSpan.Zero)) msg.Close(); } } From 5d23edc07f72a02c7dd3f61c94dec7fc891a95b4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 24 May 2026 04:09:28 +0000 Subject: [PATCH 5/6] fix: catch TerminatingException in reader task and restore positive timeout to stress the race Agent-Logs-Url: https://github.com/zeromq/netmq/sessions/6a257730-950c-4974-a634-78c12e549d5b Co-authored-by: drewnoakes <350947+drewnoakes@users.noreply.github.com> --- .../ThreadSafeSocketPollerCleanupTests.cs | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs b/src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs index 36603777..77a1086a 100644 --- a/src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs +++ b/src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs @@ -25,12 +25,18 @@ public class ThreadSafeSocketPollerCleanupTests : IClassFixture /// Repeatedly creates a , connects several - /// s to it, sends messages concurrently from multiple - /// threads (to keep the user thread processing commands), then disposes the server - /// while new clients are still connecting. + /// s to it, reads from it with a small positive timeout + /// (so the reader thread holds m_threadSafeSync while blocked in + /// ProcessCommands), then disposes the server while new clients are still + /// connecting. /// - /// Without the fix this reliably throws ArgumentOutOfRangeException inside - /// SocketBase.ProcessTerm when run enough iterations. + /// Without the fix this reliably throws + /// inside SocketBase.ProcessTerm when run enough iterations. + /// + /// The reader task also catches because + /// ProcessCommands calls CheckContextTerminated() after draining the + /// mailbox, and that check can fire once the reaper has set m_isStopped on + /// the socket — a normal part of socket shutdown that the reader must tolerate. /// [Fact] public async Task ClosingServerWhileClientsConnectDoesNotCrash() @@ -56,12 +62,13 @@ public async Task ClosingServerWhileClientsConnectDoesNotCrash() msg.InitEmpty(); while (!cts.IsCancellationRequested) { - if (server.TryReceive(ref msg, TimeSpan.Zero)) + if (server.TryReceive(ref msg, TimeSpan.FromMilliseconds(1))) msg.Close(); } } catch (ObjectDisposedException) { } catch (OperationCanceledException) { } + catch (TerminatingException) { } }); // Connect several clients. Each successful TCP handshake will enqueue a From 0cc9b8cfb1c22521b12f12b8bc278b81296dd50a Mon Sep 17 00:00:00 2001 From: Drew Noakes Date: Sun, 24 May 2026 15:03:10 +1000 Subject: [PATCH 6/6] Use nullable restore, not disable --- src/NetMQ/Core/Transports/StreamEngine.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/NetMQ/Core/Transports/StreamEngine.cs b/src/NetMQ/Core/Transports/StreamEngine.cs index 31f2128d..e41e01ef 100644 --- a/src/NetMQ/Core/Transports/StreamEngine.cs +++ b/src/NetMQ/Core/Transports/StreamEngine.cs @@ -1222,7 +1222,7 @@ void MechanismReady () m_nextMsg = PullAndEncode; m_processMsg = DecodeAndPush; } -#nullable disable +#nullable restore PullMsgResult PullAndEncode (ref Msg msg) {