From 9b3f32ce24f0a79a9d6de1f647c7c39210d69435 Mon Sep 17 00:00:00 2001
From: Michael Alt <m.alt@rational-online.com>
Date: Wed, 2 Sep 2026 13:45:14 +0200
Subject: [PATCH] Listener: defer Open frame until peer protocol header is
 received

The listener connection pipelines its AMQP protocol header and Open frame
immediately after SASL (OpenPipe state). When both are flushed into a single
TCP segment, some clients fail to parse the coalesced bytes and hang until
their connection open timeout. This is intermittent and timing dependent
(it depends on how the bytes are chunked on the wire), and is reproducible
with the Node @azure/service-bus / rhea client opening several connections
concurrently against a ConnectionListener.

Defer the listener's Open until the peer's protocol header arrives, so the
header and Open are sent as separate writes. This matches the more
conservative handshake used by brokers such as Azure Service Bus and is
transparent to clients that already handle the pipelined case.

A new internal virtual Connection.PipelineOpen (default true) controls the
behaviour; ListenerConnection overrides it to false. Client connections are
unchanged.
---
 src/Connection.cs                  | 40 ++++++++++++++++++++++++++++--
 src/Listener/ListenerConnection.cs |  8 ++++++
 2 files changed, 46 insertions(+), 2 deletions(-)

diff --git a/src/Connection.cs b/src/Connection.cs
index 7f19b77b..63249264 100644
--- a/src/Connection.cs
+++ b/src/Connection.cs
@@ -121,6 +121,16 @@ public partial class Connection : AmqpObject
         uint remoteMaxFrameSize;
         ITransport writer;
         HeartBeat heartBeat;
+        Open deferredOpen;
+
+        // When false, the local Open is not pipelined with the protocol header but is
+        // deferred until the peer's protocol header is received. Listener connections
+        // set this to avoid coalescing the header and Open into a single TCP segment,
+        // which some clients (e.g. the Node rhea library) fail to parse.
+        internal virtual bool PipelineOpen
+        {
+            get { return true; }
+        }
 
         Connection(Address address, ushort channelMax, uint maxFrameSize)
         {
@@ -280,8 +290,17 @@ internal void Init(IBufferManager bufferManager, AmqpSettings amqpSettings, IAsy
             }
 
             this.SendHeader();
-            this.SendOpen(open);
-            this.state = ConnectionState.OpenPipe;
+            if (this.PipelineOpen)
+            {
+                this.SendOpen(open);
+                this.state = ConnectionState.OpenPipe;
+            }
+            else
+            {
+                // Defer the Open until the peer's protocol header arrives (see PipelineOpen).
+                this.deferredOpen = open;
+                this.state = ConnectionState.HeaderSent;
+            }
         }
 
         static ConnectionFactory connectionFactory;
@@ -517,6 +536,14 @@ protected override bool OnClose(Error error)
                 {
                     newState = ConnectionState.OpenClosePipe;
                 }
+                else if (this.state == ConnectionState.HeaderSent)
+                {
+                    // Non-pipelined Open was deferred and never sent; send it now so the
+                    // peer sees a well-formed Open/Close sequence once its header arrives.
+                    this.SendOpen(this.deferredOpen);
+                    this.deferredOpen = null;
+                    newState = ConnectionState.OpenClosePipe;
+                }
                 else if (state == ConnectionState.OpenSent)
                 {
                     newState = ConnectionState.ClosePipe;
@@ -764,6 +791,15 @@ internal bool OnHeader(ProtocolHeader header)
                 {
                     this.state = ConnectionState.OpenSent;
                 }
+                else if (this.state == ConnectionState.HeaderSent)
+                {
+                    // Deferred (non-pipelined) Open: now that the peer's header has
+                    // arrived, send our Open as a separate write so it is not coalesced
+                    // with the protocol header.
+                    this.SendOpen(this.deferredOpen);
+                    this.deferredOpen = null;
+                    this.state = ConnectionState.OpenSent;
+                }
                 else if (this.state == ConnectionState.OpenClosePipe)
                 {
                     this.state = ConnectionState.ClosePipe;
diff --git a/src/Listener/ListenerConnection.cs b/src/Listener/ListenerConnection.cs
index 631fab76..d297e92e 100644
--- a/src/Listener/ListenerConnection.cs
+++ b/src/Listener/ListenerConnection.cs
@@ -37,6 +37,14 @@ internal ListenerConnection(ConnectionListener listener, Address address, IHandl
             this.listener = listener;
         }
 
+        // Defer the local Open until the client's protocol header is received. Pipelining
+        // the header and Open into one TCP segment breaks some clients (Node rhea), which
+        // intermittently fail to parse the coalesced bytes and hang until their open timeout.
+        internal override bool PipelineOpen
+        {
+            get { return false; }
+        }
+
         /// <summary>
         /// Gets a IPrincipal object for the connection. If the value is null,
         /// the connection is not authenticated.