dropped writes at handler removal errors

unbuffering writes now flushes correctly if the buffer is flushed during
unbuffering

Author: rnro

Sources/NIOExtras/HTTP1ProxyConnectHandler.swift

@@ -92,22 +92,13 @@ public final class NIOHTTP1ProxyConnectHandler: ChannelDuplexHandler, RemovableC
             self.failWithError(.noResult(), context: context)
 
         case .completed:
-            let hadMark = self.bufferedWrittenMessages.hasMark
-            while self.bufferedWrittenMessages.hasMark {
-                // write until mark
-                let bufferedPart = self.bufferedWrittenMessages.removeFirst()
-                context.write(bufferedPart.data, promise: bufferedPart.promise)
-            }
-
-            // flush any messages up to the mark
-            if hadMark {
-                context.flush()
-            }
-
-            // write remainder
-            while let bufferedPart = self.bufferedWrittenMessages.popFirst() {
+            while let (bufferedPart, isMarked) = self.bufferedWrittenMessages.popFirstCheckMarked() {
                 context.write(bufferedPart.data, promise: bufferedPart.promise)
+                if isMarked {
+                    context.flush()
+                }
             }
+            
         }
 
         context.leavePipeline(removalToken: removalToken)
@@ -122,9 +113,12 @@ public final class NIOHTTP1ProxyConnectHandler: ChannelDuplexHandler, RemovableC
     public func handlerRemoved(context: ChannelHandlerContext) {
         switch self.state {
         case .failed, .completed:
-            // we don't expect there to be any buffered messages in these states
-            assert(self.bufferedWrittenMessages.isEmpty)
+            guard self.bufferedWrittenMessages.isEmpty else {
+                self.failWithError(Error.droppedWrites(), context: context)
+                return
+            }
             break
+
         case .initialized, .connectSent, .headReceived:
             self.failWithError(Error.noResult(), context: context)
         }
@@ -276,6 +270,7 @@ public final class NIOHTTP1ProxyConnectHandler: ChannelDuplexHandler, RemovableC
             case httpProxyHandshakeTimeout
             case noResult
             case channelUnexpectedlyInactive
+            case droppedWrites
         }
 
         final class Storage: Sendable {
@@ -331,6 +326,10 @@ public final class NIOHTTP1ProxyConnectHandler: ChannelDuplexHandler, RemovableC
             Error(error: .channelUnexpectedlyInactive, file: file, line: line)
         }
 
+        public static func droppedWrites(file: String = #file, line: UInt = #line) -> Error {
+            Error(error: .droppedWrites, file: file, line: line)
+        }
+
         fileprivate var errorCode: Int {
             switch self.store.details {
             case .proxyAuthenticationRequired:
@@ -347,6 +346,8 @@ public final class NIOHTTP1ProxyConnectHandler: ChannelDuplexHandler, RemovableC
                 return 5
             case .channelUnexpectedlyInactive:
                 return 6
+            case .droppedWrites:
+                return 7
             }
         }
     }
@@ -388,6 +389,8 @@ extension NIOHTTP1ProxyConnectHandler.Error.Details: CustomStringConvertible {
             return "No Result"
         case .channelUnexpectedlyInactive:
             return "Channel Unexpectedly Inactive"
+        case .droppedWrites:
+            return "Handler Was Removed with Writes Left in the Buffer"
         }
     }
 }

Sources/NIOExtras/MarkedCircularBuffer+PopFirstCheckMarked.swift

@@ -0,0 +1,23 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) YEARS Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+
+extension MarkedCircularBuffer {
+    @inlinable
+    internal mutating func popFirstCheckMarked() -> (Element, Bool)? {
+        let marked = self.markedElementIndex == self.startIndex
+        return self.popFirst().map { ($0, marked) }
+    }
+}