Tasks returned by grpc stub methods are not idempotent

[Jasper-M]

We have an interceptor that works like this:

case class CredentialsSetter(credentials: CallCredentials) extends ClientInterceptor() {
  override def interceptCall[ReqT, RespT](
      method: MethodDescriptor[ReqT, RespT],
      callOptions: CallOptions,
      next: Channel
    ): ClientCall[ReqT, RespT] = {
    next.newCall(method, callOptions.withCallCredentials(credentials))
  }
}

case class StaticTokenCredentials(token: String) extends CallCredentials {
  override def applyRequestMetadata(
      requestInfo: CallCredentials.RequestInfo,
      appExecutor: Executor,
      applier: CallCredentials.MetadataApplier
    ): Unit = {
    val requestHeaders = new Metadata().withToken(token)
    applier.apply(requestHeaders)
  }
  override def thisUsesUnstableApi(): Unit = {}
}

Now when we have a stub service with this interceptor applied to its channel:

(service.someGrpcCall(Empty()) <* Task.sleep(1.minute))
  .flatMap(response => Task(println(s"received $response")))
  .restartUntil(_ => false)
  .runToFuture

Every subsequent call to someGrpcCall will contain one extra token header until the header grows so big that you get an exception Header size exceeded max allowed size (8192).

The reason is that here in grpc-java the origHeaders is a mutable object and our requestHeaders above are getting merged into it. And in the stub code we have ClientCall(channel, MyServiceApi.METHOD_SOME_GRPC_CALL, opts).unaryToUnaryCall(request, metadata). The metadata here gets reused on every subsequent call, and every time the MetadataApplier mutates it.

I think the stub implementation should be something like this to make it safe:

Task.defer{
  val metadataCopy = new Metadata().merge(metadata)
  ClientCall(channel, MyServiceApi.METHOD_SOME_GRPC_CALL, opts).unaryToUnaryCall(request, metadataCopy)
}