diff --git a/src/StackExchange.Redis/PhysicalConnection.Transport.cs b/src/StackExchange.Redis/PhysicalConnection.Transport.cs index a0258166a..9b7a53dc9 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Transport.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Transport.cs @@ -37,8 +37,17 @@ private void InitTransportOutput(DuplexTransport transport) #endif } + private bool _transportReadingStarted; + + /// Attach the receiver and begin inbound delivery. Idempotent because it is now called + /// from TWO places: as soon as the transport is adopted (before the handshake is written, which is + /// the point), and from , which every connection still goes through and + /// which must not start a second receiver. private void StartTransportReading(DuplexTransport transport) { + if (_transportReadingStarted) return; + _transportReadingStarted = true; + _readStatus = ReadStatus.Init; _readState = default; _readBuffer = CycleBuffer.Create(pool: ReaderBufferPool); diff --git a/src/StackExchange.Redis/PhysicalConnection.cs b/src/StackExchange.Redis/PhysicalConnection.cs index 0d93eefeb..17caff418 100644 --- a/src/StackExchange.Redis/PhysicalConnection.cs +++ b/src/StackExchange.Redis/PhysicalConnection.cs @@ -184,7 +184,23 @@ internal async Task BeginConnectAsync(ILogger? log) // connectTo=null no-socket pattern): connect and TLS belong to the tunnel, and the // stream/SslStream machinery below never runs. The TLS intent goes with it precisely // because TLS is now the tunnel's job: it cannot honour an intent it cannot see. - var transport = await tunnel.ConnectTransportAsync(endpoint, bridge.ConnectionType, new(rawConfig), CancellationToken.None).ForAwait(); + RESPite.Transports.DuplexTransport? transport; + try + { + transport = await tunnel.ConnectTransportAsync(endpoint, bridge.ConnectionType, new(rawConfig), CancellationToken.None).ForAwait(); + } + catch (Exception ex) + { + // A tunnel refuses a dial by throwing, and the message is the useful part: it is where + // "this configuration asks for something this transport cannot do" gets said. This + // method runs fire-and-forget (PhysicalBridge calls it as BeginConnectAsync(log) + // .RedisFireAndForget()) and the try below has not been entered yet, so an escaping + // exception was reported nowhere - the caller saw a bare connect timeout, and the + // reason was lost. + RecordConnectionFailed(ConnectionFailureType.UnableToConnect, ex, isInitialConnect: true); + return; + } + if (transport is not null) { _transport = transport; @@ -1053,6 +1069,18 @@ internal async ValueTask ConnectedAsync(Socket? socket, ILogger? log) } InitTransportOutput(transport); + + // Start inbound delivery BEFORE anything is written. OnConnectedAsync below sends the + // handshake, and a transport is already connected by the time we get here, so if the + // receiver were attached afterwards (as it was until now, via StartReading once this + // method returned) the reply could be on the wire first. That is not theoretical: it + // cost every SUBSCRIBE on a transport-backed multiplexer, because the subscription + // connection lost that race every time while the interactive one happened to win it. + // Starting first also makes DuplexTransport.Start's contract - a receiver set "before + // any data is expected" - true, rather than something each implementer must discover + // and work around by buffering. + StartTransportReading(transport); + log?.LogInformationTransportConnected(bridge.Name, transport.IsEncrypted); await bridge.OnConnectedAsync(this, log).ForAwait(); return true;