Enable Swift 6 language mode for a number of v2 modules (#1959)

Enable Swift 6 language mode for a number of v2 modules

Motivation:

v2 will eventually require the v6 language, doing this in one go is a
reasonably large change so adopt a few targets at a time.

Modifications:

- Fixup warnings/errors in the following targets:
  - GRPCInterceptors
  - GRPCInProcessTransport
  - GRPCHTTP2Core
  - GRPCHTTP2TransportNIOPosix
  - GRPCHTTP2TransportNIOTransportServices
  - GRPCCodeGen
  - GRPCProtobufCodeGenTests
  - GRPCProtobufCodeGen
  - GRPCProtobufTests
  - GRPCProtobuf
  - interoperability-tests
  - performance-worker
- Fixup warnings/errors but remain in the v5 language mode for the
  following:
  - GRPCInterceptorsTests

Result:

More modules use v6 language mode

Sources/GRPCHTTP2Core/Client/Connection/ClientConnectionHandler.swift

@@ -198,9 +198,10 @@ final class ClientConnectionHandler: ChannelInboundHandler, ChannelOutboundHandl
       // Pings are ack'd by the HTTP/2 handler so we only pay attention to acks here, and in
       // particular only those carrying the keep-alive data.
       if ack, data == self.keepalivePingData {
+        let loopBound = LoopBoundView(handler: self, context: context)
         self.keepaliveTimeoutTimer.cancel()
         self.keepaliveTimer?.schedule(on: context.eventLoop) {
-          self.keepaliveTimerFired(context: context)
+          loopBound.keepaliveTimerFired()
         }
       }
 
@@ -211,12 +212,13 @@ final class ClientConnectionHandler: ChannelInboundHandler, ChannelOutboundHandl
       // becoming active is insufficient as, for example, a TLS handshake may fail after
       // establishing the TCP connection, or the server isn't configured for gRPC (or HTTP/2).
       if isInitialSettings {
+        let loopBound = LoopBoundView(handler: self, context: context)
         self.keepaliveTimer?.schedule(on: context.eventLoop) {
-          self.keepaliveTimerFired(context: context)
+          loopBound.keepaliveTimerFired()
         }
 
         self.maxIdleTimer?.schedule(on: context.eventLoop) {
-          self.maxIdleTimerFired(context: context)
+          loopBound.maxIdleTimerFired()
         }
 
         context.fireChannelRead(self.wrapInboundOut(.ready))
@@ -264,6 +266,33 @@ final class ClientConnectionHandler: ChannelInboundHandler, ChannelOutboundHandl
   }
 }
 
+extension ClientConnectionHandler {
+  struct LoopBoundView: @unchecked Sendable {
+    private let handler: ClientConnectionHandler
+    private let context: ChannelHandlerContext
+
+    init(handler: ClientConnectionHandler, context: ChannelHandlerContext) {
+      self.handler = handler
+      self.context = context
+    }
+
+    func keepaliveTimerFired() {
+      self.context.eventLoop.assertInEventLoop()
+      self.handler.keepaliveTimerFired(context: self.context)
+    }
+
+    func keepaliveTimeoutExpired() {
+      self.context.eventLoop.assertInEventLoop()
+      self.handler.keepaliveTimeoutExpired(context: self.context)
+    }
+
+    func maxIdleTimerFired() {
+      self.context.eventLoop.assertInEventLoop()
+      self.handler.maxIdleTimerFired(context: self.context)
+    }
+  }
+}
+
 extension ClientConnectionHandler {
   struct HTTP2StreamDelegate: @unchecked Sendable, NIOHTTP2StreamDelegate {
     // @unchecked is okay: the only methods do the appropriate event-loop dance.
@@ -315,8 +344,9 @@ extension ClientConnectionHandler {
     case .startIdleTimer(let cancelKeepalive):
       // All streams are closed, restart the idle timer, and stop the keep-alive timer (it may
       // not stop if keep-alive is allowed when there are no active calls).
+      let loopBound = LoopBoundView(handler: self, context: context)
       self.maxIdleTimer?.schedule(on: context.eventLoop) {
-        self.maxIdleTimerFired(context: context)
+        loopBound.maxIdleTimerFired()
       }
 
       if cancelKeepalive {
@@ -355,8 +385,9 @@ extension ClientConnectionHandler {
     self.maybeFlush(context: context)
 
     // Schedule a timeout on waiting for the response.
+    let loopBound = LoopBoundView(handler: self, context: context)
     self.keepaliveTimeoutTimer.schedule(on: context.eventLoop) {
-      self.keepaliveTimeoutExpired(context: context)
+      loopBound.keepaliveTimeoutExpired()
     }
   }
 

Sources/GRPCHTTP2Core/Client/Connection/ConnectionFactory.swift

@@ -26,7 +26,7 @@ public protocol HTTP2Connector: Sendable {
 
 @_spi(Package)
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
-public struct HTTP2Connection {
+public struct HTTP2Connection: Sendable {
   /// The underlying TCP connection wrapped up for use with gRPC.
   var channel: NIOAsyncChannel<ClientConnectionEvent, Void>
 

Sources/GRPCHTTP2Core/Client/Connection/GRPCChannel.swift

@@ -189,7 +189,7 @@ public struct GRPCChannel: ClientTransport {
   }
 
   /// Opens a stream using the transport, and uses it as input into a user-provided closure.
-  public func withStream<T>(
+  public func withStream<T: Sendable>(
     descriptor: MethodDescriptor,
     options: CallOptions,
     _ closure: (_ stream: RPCStream<Inbound, Outbound>) async throws -> T

Sources/GRPCHTTP2Core/Internal/ConstantAsyncSequence.swift

@@ -17,7 +17,7 @@
 import GRPCCore
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
-private struct ConstantAsyncSequence<Element: Sendable>: AsyncSequence {
+private struct ConstantAsyncSequence<Element: Sendable>: AsyncSequence, Sendable {
   private let element: Element
 
   init(element: Element) {
@@ -42,7 +42,7 @@ private struct ConstantAsyncSequence<Element: Sendable>: AsyncSequence {
 }
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
-extension RPCAsyncSequence {
+extension RPCAsyncSequence where Element: Sendable {
   static func constant(_ element: Element) -> RPCAsyncSequence<Element> {
     return RPCAsyncSequence(wrapping: ConstantAsyncSequence(element: element))
   }

Sources/GRPCHTTP2Core/Internal/DiscardingTaskGroup+CancellableHandle.swift

@@ -55,7 +55,7 @@ extension DiscardingTaskGroup {
   }
 
   @usableFromInline
-  enum FinishedOrCancelled {
+  enum FinishedOrCancelled: Sendable {
     case finished
     case cancelled
   }

Sources/GRPCHTTP2Core/Internal/Timer.swift

@@ -45,7 +45,7 @@ struct Timer {
   }
 
   /// Schedule a task on the given `EventLoop`.
-  mutating func schedule(on eventLoop: EventLoop, work: @escaping () throws -> Void) {
+  mutating func schedule(on eventLoop: EventLoop, work: @escaping @Sendable () throws -> Void) {
     self.task?.cancel()
 
     if self.repeat {

Sources/GRPCHTTP2Core/Server/Connection/ServerConnectionManagementHandler.swift

@@ -247,16 +247,18 @@ final class ServerConnectionManagementHandler: ChannelDuplexHandler {
   }
 
   func channelActive(context: ChannelHandlerContext) {
+    let view = LoopBoundView(handler: self, context: context)
+
     self.maxAgeTimer?.schedule(on: context.eventLoop) {
-      self.initiateGracefulShutdown(context: context)
+      view.initiateGracefulShutdown()
     }
 
     self.maxIdleTimer?.schedule(on: context.eventLoop) {
-      self.initiateGracefulShutdown(context: context)
+      view.initiateGracefulShutdown()
     }
 
     self.keepaliveTimer?.schedule(on: context.eventLoop) {
-      self.keepaliveTimerFired(context: context)
+      view.keepaliveTimerFired()
     }
 
     context.fireChannelActive()
@@ -321,8 +323,9 @@ final class ServerConnectionManagementHandler: ChannelDuplexHandler {
     self.inReadLoop = false
 
     // Done reading: schedule the keep-alive timer.
+    let view = LoopBoundView(handler: self, context: context)
     self.keepaliveTimer?.schedule(on: context.eventLoop) {
-      self.keepaliveTimerFired(context: context)
+      view.keepaliveTimerFired()
     }
 
     context.fireChannelReadComplete()
@@ -333,6 +336,29 @@ final class ServerConnectionManagementHandler: ChannelDuplexHandler {
   }
 }
 
+extension ServerConnectionManagementHandler {
+  struct LoopBoundView: @unchecked Sendable {
+    private let handler: ServerConnectionManagementHandler
+    private let context: ChannelHandlerContext
+
+    init(handler: ServerConnectionManagementHandler, context: ChannelHandlerContext) {
+      self.handler = handler
+      self.context = context
+    }
+
+    func initiateGracefulShutdown() {
+      self.context.eventLoop.assertInEventLoop()
+      self.handler.initiateGracefulShutdown(context: self.context)
+    }
+
+    func keepaliveTimerFired() {
+      self.context.eventLoop.assertInEventLoop()
+      self.handler.keepaliveTimerFired(context: self.context)
+    }
+
+  }
+}
+
 extension ServerConnectionManagementHandler {
   struct HTTP2StreamDelegate: @unchecked Sendable, NIOHTTP2StreamDelegate {
     // @unchecked is okay: the only methods do the appropriate event-loop dance.
@@ -379,8 +405,9 @@ extension ServerConnectionManagementHandler {
 
     switch self.state.streamClosed(id) {
     case .startIdleTimer:
+      let loopBound = LoopBoundView(handler: self, context: context)
       self.maxIdleTimer?.schedule(on: context.eventLoop) {
-        self.initiateGracefulShutdown(context: context)
+        loopBound.initiateGracefulShutdown()
       }
 
     case .close:
@@ -481,8 +508,9 @@ extension ServerConnectionManagementHandler {
       } else {
         // RPCs may have a grace period for finishing once the second GOAWAY frame has finished.
         // If this is set close the connection abruptly once the grace period passes.
+        let loopBound = NIOLoopBound(context, eventLoop: context.eventLoop)
         self.maxGraceTimer?.schedule(on: context.eventLoop) {
-          context.close(promise: nil)
+          loopBound.value.close(promise: nil)
         }
       }
 
@@ -497,8 +525,9 @@ extension ServerConnectionManagementHandler {
     self.maybeFlush(context: context)
 
     // Schedule a timeout on waiting for the response.
+    let loopBound = LoopBoundView(handler: self, context: context)
     self.keepaliveTimeoutTimer.schedule(on: context.eventLoop) {
-      self.initiateGracefulShutdown(context: context)
+      loopBound.initiateGracefulShutdown()
     }
   }
 }

Sources/GRPCHTTP2TransportNIOPosix/HTTP2ClientTransport+Posix.swift

@@ -109,7 +109,7 @@ extension HTTP2ClientTransport {
       self.channel.close()
     }
 
-    public func withStream<T>(
+    public func withStream<T: Sendable>(
       descriptor: MethodDescriptor,
       options: CallOptions,
       _ closure: (RPCStream<Inbound, Outbound>) async throws -> T

Sources/GRPCHTTP2TransportNIOPosix/HTTP2ServerTransport+Posix.swift

@@ -162,7 +162,7 @@ extension HTTP2ServerTransport {
     }
 
     public func listen(
-      _ streamHandler: @escaping (RPCStream<Inbound, Outbound>) async -> Void
+      _ streamHandler: @escaping @Sendable (RPCStream<Inbound, Outbound>) async -> Void
     ) async throws {
       defer {
         switch self.listeningAddressState.withLockedValue({ $0.close() }) {

Sources/GRPCHTTP2TransportNIOPosix/NIOClientBootstrap+SocketAddress.swift

@@ -21,7 +21,7 @@ import NIOPosix
 
 extension ClientBootstrap {
   @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
-  func connect<Result>(
+  func connect<Result: Sendable>(
     to address: GRPCHTTP2Core.SocketAddress,
     _ configure: @Sendable @escaping (Channel) -> EventLoopFuture<Result>
   ) async throws -> Result {

Sources/GRPCHTTP2TransportNIOTransportServices/HTTP2ServerTransport+TransportServices.swift

@@ -156,7 +156,7 @@ extension HTTP2ServerTransport {
     }
 
     public func listen(
-      _ streamHandler: @escaping (RPCStream<Inbound, Outbound>) async -> Void
+      _ streamHandler: @escaping @Sendable (RPCStream<Inbound, Outbound>) async -> Void
     ) async throws {
       defer {
         switch self.listeningAddressState.withLockedValue({ $0.close() }) {