Upgrade grpc-swift-nio-transport to 2.9.0 and remove HTTP2ConnectBuff… (#1790)

- Fixes #1789.
- Release 2.9.0 of `grpc-swift-nio-transport` fixes
  an HTTP/2 initialization race where the server could
  send SETTINGS before gRPC handlers are added to
  the pipeline, causing the client to hang. The new
  `WrappedChannel.wrapping(config:serviceConfig:makeChannel:)`
  API calls `configure(channel)` inside the channel
  initializer, ensuring the pipeline is set up before any inbound
  bytes arrive. This eliminates the need for the custom
  `HTTP2ConnectBufferingHandler` workaround.

Author: adityabagchi24

Package.resolved

@@ -62,7 +62,7 @@ let package = Package(
         .package(url: "https://github.com/apple/swift-protobuf.git", from: "1.36.0"),
         .package(url: "https://github.com/apple/swift-system.git", from: "1.6.4"),
         .package(url: "https://github.com/grpc/grpc-swift-2.git", from: "2.3.0"),
-        .package(url: "https://github.com/grpc/grpc-swift-nio-transport.git", from: "2.4.4"),
+        .package(url: "https://github.com/grpc/grpc-swift-nio-transport.git", from: "2.9.0"),
         .package(url: "https://github.com/grpc/grpc-swift-protobuf.git", from: "2.2.0"),
         .package(url: "https://github.com/swift-server/async-http-client.git", from: "1.20.1"),
         .package(url: "https://github.com/swiftlang/swift-docc-plugin.git", from: "1.1.0"),

Package.swift

@@ -24,9 +24,6 @@ import GRPCCore
 import GRPCNIOTransportHTTP2
 import Logging
 import NIO
-import NIOCore
-import NIOHPACK
-import NIOHTTP2
 import NIOPosix
 
 public struct Builder: Sendable {
@@ -39,22 +36,27 @@ public struct Builder: Sendable {
     let clientTask: Task<Void, any Swift.Error>
     let logger: Logger
 
-    public init(socket: FileHandle, group: EventLoopGroup, logger: Logger) throws {
+    public init(socket: FileHandle, group: EventLoopGroup, logger: Logger) async throws {
         try socket.setSendBufSize(4 << 20)
         try socket.setRecvBufSize(2 << 20)
 
-        let channel = try ClientBootstrap(group: group)
-            .channelInitializer { channel in
-                channel.eventLoop.makeCompletedFuture(withResultOf: {
-                    try channel.pipeline.syncOperations.addHandler(HTTP2ConnectBufferingHandler())
-                })
+        let transport = try await HTTP2ClientTransport.WrappedChannel.wrapping(
+            config: .defaults,
+            serviceConfig: .init()
+        ) { configure in
+            try await withCheckedThrowingContinuation { continuation in
+                ClientBootstrap(group: group)
+                    .channelInitializer { channel in
+                        configure(channel).map { configured in
+                            continuation.resume(returning: configured)
+                        }
+                    }
+                    .withConnectedSocket(socket.fileDescriptor)
+                    .whenFailure { error in
+                        continuation.resume(throwing: error)
+                    }
             }
-            .withConnectedSocket(socket.fileDescriptor)
-            .wait()
-
-        let transport = HTTP2ClientTransport.WrappedChannel.wrapping(
-            channel: channel
-        )
+        }
 
         let grpcClient = GRPCClient(transport: transport)
         self.grpcClient = grpcClient
@@ -429,49 +431,3 @@ extension FileHandle {
         }
     }
 }
-
-/// Buffers incoming bytes until the full gRPC HTTP/2 pipeline is configured, then replays them.
-///
-/// See the equivalent in Containerization/Vminitd.swift for a full explanation.
-private final class HTTP2ConnectBufferingHandler: ChannelDuplexHandler, RemovableChannelHandler {
-    typealias InboundIn = ByteBuffer
-    typealias InboundOut = ByteBuffer
-    typealias OutboundIn = ByteBuffer
-    typealias OutboundOut = ByteBuffer
-
-    private var removalScheduled = false
-    private var bufferedReads: [NIOAny] = []
-
-    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
-        bufferedReads.append(data)
-    }
-
-    func channelReadComplete(context: ChannelHandlerContext) {}
-
-    func flush(context: ChannelHandlerContext) {
-        if !removalScheduled {
-            removalScheduled = true
-            context.eventLoop.assumeIsolatedUnsafeUnchecked().execute {
-                context.pipeline.syncOperations.removeHandler(self, promise: nil)
-            }
-        }
-        context.flush()
-    }
-
-    func removeHandler(context: ChannelHandlerContext, removalToken: ChannelHandlerContext.RemovalToken) {
-        var didRead = false
-        while !bufferedReads.isEmpty {
-            context.fireChannelRead(bufferedReads.removeFirst())
-            didRead = true
-        }
-        if didRead {
-            context.fireChannelReadComplete()
-        }
-        context.leavePipeline(removalToken: removalToken)
-    }
-
-    func channelInactive(context: ChannelHandlerContext) {
-        bufferedReads.removeAll()
-        context.fireChannelInactive()
-    }
-}

Sources/ContainerBuild/Builder.swift

@@ -177,7 +177,7 @@ extension Application {
                                 let fh = try await client.dial(id: "buildkit", port: vsockPort)
 
                                 let threadGroup: MultiThreadedEventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: System.coreCount)
-                                let b = try Builder(socket: fh, group: threadGroup, logger: log)
+                                let b = try await Builder(socket: fh, group: threadGroup, logger: log)
 
                                 // If this call succeeds, then BuildKit is running.
                                 let _ = try await b.info()