fix: minor fixes

better release

Author: st0o0

src/Servus.Akka/Transport/Quic/Client/QuicTransportStateMachine.cs

@@ -71,7 +71,15 @@ internal void Dispatch(IQuicTransportEvent evt)
 
                 break;
             case InboundPumpFailed e:
-                OnInboundComplete(DisconnectReason.Error, e.StreamId);
+                if (IsConnectionLevelError(e.Error))
+                {
+                    HandleConnectionFailure(DisconnectReason.Error);
+                }
+                else
+                {
+                    OnInboundComplete(DisconnectReason.Error, e.StreamId);
+                }
+
                 break;
             case OutboundWriteDone:
                 _ops.OnSignalPullOutbound();
@@ -313,29 +321,30 @@ private void OnAcquisitionFailed(Exception ex)
         _ops.OnCancelTimer(ConnectTimerKey);
         Tracing.For("Connection").Warning(this, "QUIC acquisition failed: {0}", ex.Message);
 
-        if (_pendingConnect is null)
+        if (_pendingConnect is not null)
         {
+            _pendingConnect = null;
+            _ops.OnPushInbound(new TransportDisconnected(DisconnectReason.Error));
+            _ops.OnSignalPullOutbound();
             return;
         }
 
-        _pendingConnect = null;
-        _ops.OnPushInbound(new TransportDisconnected(DisconnectReason.Error));
-        _ops.OnSignalPullOutbound();
+        HandleConnectionFailure(DisconnectReason.Error);
     }
 
     private void HandleConnectionFailure(DisconnectReason reason)
     {
         Tracing.For("Connection").Debug(this, "QUIC disconnected: {0}", reason);
-        foreach (var (streamId, state) in _streams)
-        {
-            _ops.OnPushInbound(new StreamClosed(streamId, reason));
-            _ = state.DisposeAsync();
-        }
-
-        _streams.Clear();
 
         if (_autoReconnect && !_upstreamFinished)
         {
+            foreach (var (_, state) in _streams)
+            {
+                _ = state.DisposeAsync();
+            }
+
+            _streams.Clear();
+
             _ops.OnPushInbound(new TransportDisconnected(DisconnectReason.Transient));
             _isReconnecting = true;
             _pumpManager?.StopAll();
@@ -346,6 +355,14 @@ private void HandleConnectionFailure(DisconnectReason reason)
             return;
         }
 
+        foreach (var (streamId, state) in _streams)
+        {
+            _ops.OnPushInbound(new StreamClosed(streamId, reason));
+            _ = state.DisposeAsync();
+        }
+
+        _streams.Clear();
+
         _ops.OnPushInbound(new TransportDisconnected(reason));
         _pumpManager?.StopAll();
         ReturnConnectionToPool(false);
@@ -439,6 +456,18 @@ private void CleanupTransport()
         _connectionHandle = null;
         _connectionLease = null;
     }
+    private static bool IsConnectionLevelError(Exception ex)
+    {
+        if (ex is System.Net.Quic.QuicException qe)
+        {
+            return qe.QuicError is System.Net.Quic.QuicError.ConnectionAborted
+                or System.Net.Quic.QuicError.ConnectionIdle
+                or System.Net.Quic.QuicError.ConnectionRefused
+                or System.Net.Quic.QuicError.ConnectionTimeout;
+        }
+
+        return ex is ObjectDisposedException;
+    }
 }
 
 #pragma warning restore CA1416

src/TurboHTTP/Internal/OptionsFactory.cs

@@ -39,7 +39,8 @@ internal static TransportOptions Build(RequestEndpoint endpoint, TurboClientOpti
                 AllowConnectionMigration = clientOptions.Http3.AllowConnectionMigration,
                 AllowEarlyData = clientOptions.Http3.AllowEarlyData,
                 ApplicationProtocols = alpn,
-                AutoReconnect = true
+                AutoReconnect = true,
+                ConnectionLifetime = clientOptions.PooledConnectionLifetime
             };
         }
 

src/TurboHTTP/Protocol/Http3/StateMachine.cs

@@ -69,6 +69,13 @@ internal sealed class StateMachine : IDisposable
     /// <summary>Whether there are in-flight requests awaiting responses.</summary>
     public bool HasInFlightRequests => _streamManager.HasInFlightRequests;
 
+    /// <summary>
+    /// Fails an in-flight request on the given stream due to a transport error.
+    /// Returns true if a correlated request was found and failed.
+    /// </summary>
+    public bool FailInflightRequest(long streamId, Exception exception) =>
+        _streamManager.FailInflightRequest(streamId, exception);
+
     /// <summary>The current connection endpoint.</summary>
     public RequestEndpoint Endpoint { get; private set; }
 
@@ -392,6 +399,14 @@ public void OnConnectionLost()
         TableSync.Reset();
         _serverStreamMap.Clear();
         _pendingStreamType.Clear();
+
+        if (_transportOptions is not null)
+        {
+            _ops.OnOutbound(new OpenStream(-2, StreamDirection.Unidirectional));
+            _ops.OnOutbound(new OpenStream(-3, StreamDirection.Unidirectional));
+            _ops.OnOutbound(new OpenStream(-4, StreamDirection.Unidirectional));
+            _ops.OnOutbound(new ConnectTransport(_transportOptions));
+        }
     }
 
     /// <summary>
@@ -535,6 +550,11 @@ private void ReplayBufferedFrames()
         }
 
         ArrayPool<Http3Frame>.Shared.Return(replayArray, true);
+
+        for (var i = correlationIndex; i < oldCorrelations.Count; i++)
+        {
+            EncodeAndEmit(oldCorrelations[i]);
+        }
     }
 
     private void EmitSerializedFrame(Http3Frame frame, long streamId = -1)

src/TurboHTTP/Protocol/Http3/StreamManager.cs

@@ -95,6 +95,41 @@ public void FlushPendingResponse(long streamId)
         }
     }
 
+    /// <summary>
+    /// Fails an in-flight request on the given stream due to a transport error.
+    /// Removes the correlation and stream state, and completes the <see cref="PendingRequest"/>
+    /// with an exception so the caller's <c>SendAsync</c> or <c>ReadAsStringAsync</c> throws.
+    /// Returns true if a correlated request was found and failed.
+    /// </summary>
+    public bool FailInflightRequest(long streamId, Exception exception)
+    {
+        if (_streams.TryGetValue(streamId, out var state))
+        {
+            state.Reset();
+            if (_statePool.Count < MaxPoolSize)
+            {
+                _statePool.Push(state);
+            }
+
+            _streams.Remove(streamId);
+        }
+
+        if (!_correlationMap.Remove(streamId, out var request))
+        {
+            return false;
+        }
+
+        OnStreamClosedCallback?.Invoke(streamId);
+        ReturnDecoder(streamId);
+
+        if (request.Options.TryGetValue(TcsCorrelation.Key, out var pending))
+        {
+            pending.TrySetException(exception);
+        }
+
+        return true;
+    }
+
     /// <summary>
     /// Completes all in-progress response assemblies (upstream finish / connection close).
     /// </summary>

src/TurboHTTP/Streams/Stages/Http30ConnectionStage.cs

@@ -217,7 +217,16 @@ private void HandleSignalItem(ITransportInbound item)
                 }
                 case StreamClosed { StreamId: >= 0 } streamClosed:
                 {
-                    _sm.FlushPendingResponse(streamClosed.StreamId);
+                    if (streamClosed.Reason == DisconnectReason.Error)
+                    {
+                        _sm.FailInflightRequest(streamClosed.StreamId,
+                            new HttpRequestException("HTTP/3 stream aborted by transport."));
+                    }
+                    else
+                    {
+                        _sm.FlushPendingResponse(streamClosed.StreamId);
+                    }
+
                     FlushResponses();
                     TryPullRequest();
                     return;