Suspend request stream writes before the RPC is ready

.github/workflows/ci.yaml

@@ -25,7 +25,8 @@ jobs:
       matrix:
         include:
           - image: swift:5.6-focal
-            swift-test-flags: "--sanitize=thread"
+            # No TSAN because of: https://github.com/apple/swift/issues/59068
+            # swift-test-flags: "--sanitize=thread"
           - image: swift:5.5-focal
             swift-test-flags: "--sanitize=thread"
           - image: swift:5.4-focal

Sources/GRPC/AsyncAwaitSupport/AsyncWriter.swift

@@ -109,7 +109,7 @@ internal final actor AsyncWriter<Delegate: AsyncWriterDelegate>: Sendable {
 
   /// Whether the writer is paused.
   @usableFromInline
-  internal var _isPaused: Bool = false
+  internal var _isPaused: Bool
 
   /// The delegate to process elements. By convention we call the delegate before resuming any
   /// continuation.
@@ -120,12 +120,14 @@ internal final actor AsyncWriter<Delegate: AsyncWriterDelegate>: Sendable {
   internal init(
     maxPendingElements: Int = 16,
     maxWritesBeforeYield: Int = 5,
+    isWritable: Bool = true,
     delegate: Delegate
   ) {
     self._maxPendingElements = maxPendingElements
     self._maxWritesBeforeYield = maxWritesBeforeYield
     self._pendingElements = CircularBuffer(initialCapacity: maxPendingElements)
     self._completionState = .incomplete
+    self._isPaused = !isWritable
     self._delegate = delegate
   }
 

Sources/GRPC/AsyncAwaitSupport/Call+AsyncRequestStreamWriter.swift

@@ -26,7 +26,8 @@ extension Call {
       self.send(.end, promise: nil)
     }
 
-    return GRPCAsyncRequestStreamWriter(asyncWriter: .init(delegate: delegate))
+    // Start as not-writable; writability will be toggled when the stream comes up.
+    return GRPCAsyncRequestStreamWriter(asyncWriter: .init(isWritable: false, delegate: delegate))
   }
 }
 

Sources/GRPC/AsyncAwaitSupport/GRPCAsyncBidirectionalStreamingCall.swift

@@ -86,6 +86,9 @@ public struct GRPCAsyncBidirectionalStreamingCall<Request: Sendable, Response: S
     let asyncCall = Self(call: call)
 
     asyncCall.call.invokeStreamingRequests(
+      onStart: {
+        asyncCall.requestStream.asyncWriter.toggleWritabilityAsynchronously()
+      },
       onError: { error in
         asyncCall.responseParts.handleError(error)
         asyncCall.responseSource.finish(throwing: error)

Sources/GRPC/AsyncAwaitSupport/GRPCAsyncClientStreamingCall.swift

@@ -85,6 +85,9 @@ public struct GRPCAsyncClientStreamingCall<Request: Sendable, Response: Sendable
     let asyncCall = Self(call: call)
 
     asyncCall.call.invokeStreamingRequests(
+      onStart: {
+        asyncCall.requestStream.asyncWriter.toggleWritabilityAsynchronously()
+      },
       onError: { error in
         asyncCall.responseParts.handleError(error)
         asyncCall.requestStream.asyncWriter.cancelAsynchronously()

Sources/GRPC/AsyncAwaitSupport/GRPCAsyncServerStreamingCall.swift

@@ -88,6 +88,7 @@ public struct GRPCAsyncServerStreamingCall<Request: Sendable, Response: Sendable
 
     asyncCall.call.invokeUnaryRequest(
       request,
+      onStart: {},
       onError: { error in
         asyncCall.responseParts.handleError(error)
         asyncCall.responseSource.finish(throwing: error)

Sources/GRPC/AsyncAwaitSupport/GRPCAsyncUnaryCall.swift

@@ -84,6 +84,7 @@ public struct GRPCAsyncUnaryCall<Request: Sendable, Response: Sendable>: Sendabl
     self.responseParts = UnaryResponseParts(on: call.eventLoop)
     self.call.invokeUnaryRequest(
       request,
+      onStart: {},
       onError: self.responseParts.handleError(_:),
       onResponsePart: self.responseParts.handle(_:)
     )

Sources/GRPC/ClientCalls/BidirectionalStreamingCall.swift

@@ -84,6 +84,7 @@ public struct BidirectionalStreamingCall<
 
   internal func invoke() {
     self.call.invokeStreamingRequests(
+      onStart: {},
       onError: self.responseParts.handleError(_:),
       onResponsePart: self.responseParts.handle(_:)
     )

Sources/GRPC/ClientCalls/Call.swift

@@ -123,10 +123,10 @@ public final class Call<Request, Response> {
     self.options.logger.debug("starting rpc", metadata: ["path": "\(self.path)"], source: "GRPC")
 
     if self.eventLoop.inEventLoop {
-      self._invoke(onError: onError, onResponsePart: onResponsePart)
+      self._invoke(onStart: {}, onError: onError, onResponsePart: onResponsePart)
     } else {
       self.eventLoop.execute {
-        self._invoke(onError: onError, onResponsePart: onResponsePart)
+        self._invoke(onStart: {}, onError: onError, onResponsePart: onResponsePart)
       }
     }
   }
@@ -262,6 +262,7 @@ extension Call {
   /// - Important: This *must* to be called from the `eventLoop`.
   @usableFromInline
   internal func _invoke(
+    onStart: @escaping () -> Void,
     onError: @escaping (Error) -> Void,
     onResponsePart: @escaping (GRPCClientResponsePart<Response>) -> Void
   ) {
@@ -275,6 +276,7 @@ extension Call {
         withOptions: self.options,
         onEventLoop: self.eventLoop,
         interceptedBy: self._interceptors,
+        onStart: onStart,
         onError: onError,
         onResponsePart: onResponsePart
       )
@@ -354,14 +356,25 @@ extension Call {
   @inlinable
   internal func invokeUnaryRequest(
     _ request: Request,
+    onStart: @escaping () -> Void,
     onError: @escaping (Error) -> Void,
     onResponsePart: @escaping (GRPCClientResponsePart<Response>) -> Void
   ) {
     if self.eventLoop.inEventLoop {
-      self._invokeUnaryRequest(request: request, onError: onError, onResponsePart: onResponsePart)
+      self._invokeUnaryRequest(
+        request: request,
+        onStart: onStart,
+        onError: onError,
+        onResponsePart: onResponsePart
+      )
     } else {
       self.eventLoop.execute {
-        self._invokeUnaryRequest(request: request, onError: onError, onResponsePart: onResponsePart)
+        self._invokeUnaryRequest(
+          request: request,
+          onStart: onStart,
+          onError: onError,
+          onResponsePart: onResponsePart
+        )
       }
     }
   }
@@ -373,14 +386,23 @@ extension Call {
   ///   - onResponsePart: A callback invoked for each response part received.
   @inlinable
   internal func invokeStreamingRequests(
+    onStart: @escaping () -> Void,
     onError: @escaping (Error) -> Void,
     onResponsePart: @escaping (GRPCClientResponsePart<Response>) -> Void
   ) {
     if self.eventLoop.inEventLoop {
-      self._invokeStreamingRequests(onError: onError, onResponsePart: onResponsePart)
+      self._invokeStreamingRequests(
+        onStart: onStart,
+        onError: onError,
+        onResponsePart: onResponsePart
+      )
     } else {
       self.eventLoop.execute {
-        self._invokeStreamingRequests(onError: onError, onResponsePart: onResponsePart)
+        self._invokeStreamingRequests(
+          onStart: onStart,
+          onError: onError,
+          onResponsePart: onResponsePart
+        )
       }
     }
   }
@@ -389,13 +411,14 @@ extension Call {
   @usableFromInline
   internal func _invokeUnaryRequest(
     request: Request,
+    onStart: @escaping () -> Void,
     onError: @escaping (Error) -> Void,
     onResponsePart: @escaping (GRPCClientResponsePart<Response>) -> Void
   ) {
     self.eventLoop.assertInEventLoop()
     assert(self.type == .unary || self.type == .serverStreaming)
 
-    self._invoke(onError: onError, onResponsePart: onResponsePart)
+    self._invoke(onStart: onStart, onError: onError, onResponsePart: onResponsePart)
     self._send(.metadata(self.options.customMetadata), promise: nil)
     self._send(
       .message(request, .init(compress: self.isCompressionEnabled, flush: false)),
@@ -407,13 +430,14 @@ extension Call {
   /// On-`EventLoop` implementation of `invokeStreamingRequests(_:)`.
   @usableFromInline
   internal func _invokeStreamingRequests(
+    onStart: @escaping () -> Void,
     onError: @escaping (Error) -> Void,
     onResponsePart: @escaping (GRPCClientResponsePart<Response>) -> Void
   ) {
     self.eventLoop.assertInEventLoop()
     assert(self.type == .clientStreaming || self.type == .bidirectionalStreaming)
 
-    self._invoke(onError: onError, onResponsePart: onResponsePart)
+    self._invoke(onStart: onStart, onError: onError, onResponsePart: onResponsePart)
     self._send(.metadata(self.options.customMetadata), promise: nil)
   }
 }

Sources/GRPC/ClientCalls/ClientStreamingCall.swift

@@ -84,6 +84,7 @@ public struct ClientStreamingCall<RequestPayload, ResponsePayload>: StreamingReq
 
   internal func invoke() {
     self.call.invokeStreamingRequests(
+      onStart: {},
       onError: self.responseParts.handleError(_:),
       onResponsePart: self.responseParts.handle(_:)
     )

Sources/GRPC/ClientCalls/ServerStreamingCall.swift

@@ -80,6 +80,7 @@ public struct ServerStreamingCall<RequestPayload, ResponsePayload>: ClientCall {
   internal func invoke(_ request: RequestPayload) {
     self.call.invokeUnaryRequest(
       request,
+      onStart: {},
       onError: self.responseParts.handleError(_:),
       onResponsePart: self.responseParts.handle(_:)
     )

Sources/GRPC/ClientCalls/UnaryCall.swift

@@ -84,6 +84,7 @@ public struct UnaryCall<RequestPayload, ResponsePayload>: UnaryResponseClientCal
   internal func invoke(_ request: RequestPayload) {
     self.call.invokeUnaryRequest(
       request,
+      onStart: {},
       onError: self.responseParts.handleError(_:),
       onResponsePart: self.responseParts.handle(_:)
     )

Sources/GRPC/Interceptor/ClientTransport.swift

@@ -93,6 +93,9 @@ internal final class ClientTransport<Request, Response> {
   /// The `NIO.Channel` used by the transport, if it is available.
   private var channel: Channel?
 
+  /// A callback which is invoked once when the stream channel becomes active.
+  private let onStart: () -> Void
+
   /// Our current state as logging metadata.
   private var stateForLogging: Logger.MetadataValue {
     if self.state.mayBuffer {
@@ -109,11 +112,13 @@ internal final class ClientTransport<Request, Response> {
     serializer: AnySerializer<Request>,
     deserializer: AnyDeserializer<Response>,
     errorDelegate: ClientErrorDelegate?,
+    onStart: @escaping () -> Void,
     onError: @escaping (Error) -> Void,
     onResponsePart: @escaping (GRPCClientResponsePart<Response>) -> Void
   ) {
     self.callEventLoop = eventLoop
     self.callDetails = details
+    self.onStart = onStart
     let logger = GRPCLogger(wrapping: details.options.logger)
     self.logger = logger
     self.serializer = serializer
@@ -332,6 +337,7 @@ extension ClientTransport {
       self._pipeline?.logger = self.logger
       self.logger.debug("activated stream channel")
       self.channel = channel
+      self.onStart()
       self.unbuffer()
 
     case .close:

Sources/GRPC/Interceptor/ClientTransportFactory.swift

@@ -140,6 +140,7 @@ internal struct ClientTransportFactory<Request, Response> {
     withOptions options: CallOptions,
     onEventLoop eventLoop: EventLoop,
     interceptedBy interceptors: [ClientInterceptor<Request, Response>],
+    onStart: @escaping () -> Void,
     onError: @escaping (Error) -> Void,
     onResponsePart: @escaping (GRPCClientResponsePart<Response>) -> Void
   ) -> ClientTransport<Request, Response> {
@@ -151,6 +152,7 @@ internal struct ClientTransportFactory<Request, Response> {
         withOptions: options,
         onEventLoop: eventLoop,
         interceptedBy: interceptors,
+        onStart: onStart,
         onError: onError,
         onResponsePart: onResponsePart
       )
@@ -220,6 +222,7 @@ internal struct HTTP2ClientTransportFactory<Request, Response> {
     withOptions options: CallOptions,
     onEventLoop eventLoop: EventLoop,
     interceptedBy interceptors: [ClientInterceptor<Request, Response>],
+    onStart: @escaping () -> Void,
     onError: @escaping (Error) -> Void,
     onResponsePart: @escaping (GRPCClientResponsePart<Response>) -> Void
   ) -> ClientTransport<Request, Response> {
@@ -230,6 +233,7 @@ internal struct HTTP2ClientTransportFactory<Request, Response> {
       serializer: self.serializer,
       deserializer: self.deserializer,
       errorDelegate: self.errorDelegate,
+      onStart: onStart,
       onError: onError,
       onResponsePart: onResponsePart
     )
@@ -333,6 +337,7 @@ internal struct FakeClientTransportFactory<Request, Response> {
       serializer: self.requestSerializer,
       deserializer: self.responseDeserializer,
       errorDelegate: nil,
+      onStart: {},
       onError: onError,
       onResponsePart: onResponsePart
     )