NPE in RetriableStream when ClientCallStreamObserver.cancel() races ClientCall.start()

[ziad164]

What version of gRPC-Java are you using?

1.64.0. The relevant code is unchanged in 1.69.1 and 1.83.1.

What is your environment?

Reproduced on JDK 21 (macOS). Originally found on Android (grpc-android / grpc-okhttp), where it
causes a fatal crash at meaningful volume in production.

What did you do?

Cancel a unary call through the ClientCallStreamObserver handed to
ClientResponseObserver.beforeStart(), from a different thread, while ClientCall.start() is
executing.

This is what an application does when it ties call cancellation to some external lifecycle — in our
case a Kotlin coroutine cancellation handler cancels the in-flight call when the caller goes away.

What did you expect to see?

The call cancelled, or the cancel ignored. ClientCallImpl.cancelInternal already guards against a
not-yet-started call (if (stream != null)), so cancelling before start() is safe and throws
nothing.

What did you see instead?

An NPE thrown back out of cancel() on the calling thread, wrapped as a StatusRuntimeException:

io.grpc.StatusRuntimeException: UNKNOWN: Uncaught exception in the SynchronizationContext. Re-thrown.
	at io.grpc.Status.asRuntimeException(Status.java:525)
	at io.grpc.internal.RetriableStream$1.uncaughtException(RetriableStream.java:75)
	at io.grpc.SynchronizationContext.drain(SynchronizationContext.java:96)
	at io.grpc.SynchronizationContext.execute(SynchronizationContext.java:126)
	at io.grpc.internal.RetriableStream.safeCloseMasterListener(RetriableStream.java:838)
	at io.grpc.internal.RetriableStream.cancel(RetriableStream.java:531)
	at io.grpc.internal.ClientCallImpl.cancelInternal(ClientCallImpl.java:480)
	at io.grpc.internal.ClientCallImpl.cancel(ClientCallImpl.java:454)
	at io.grpc.stub.ClientCalls$CallToStreamObserverAdapter.cancel(ClientCalls.java:431)
	...
Caused by: java.lang.NullPointerException: Cannot invoke
  "io.grpc.internal.ClientStreamListener.closed(io.grpc.Status, io.grpc.internal.ClientStreamListener$RpcProgress, io.grpc.Metadata)"
  because the return value of "io.grpc.internal.RetriableStream.access$700(io.grpc.internal.RetriableStream)" is null
	at io.grpc.internal.RetriableStream$4.run(RetriableStream.java:843)
	at io.grpc.SynchronizationContext.drain(SynchronizationContext.java:94)

Analysis

ClientCallImpl.startInternal assigns the stream at line 250 and starts it at line 285:

stream = clientStreamProvider.newStream(method, callOptions, headers, context);   // :250
...
stream.start(new ClientStreamListenerImpl(observer));                             // :285

RetriableStream.masterListener is assigned as the first statement of RetriableStream.start()
(RetriableStream.java:388). So between those two lines the stream is non-null but masterListener
is still null.

A cancel() arriving in that window passes the stream != null guard in cancelInternal, reaches
RetriableStream.cancel :531 and safeCloseMasterListener :838, whose Runnable dereferences the
null masterListener at :843. Because listenerSerializeExecutor is a SynchronizationContext
(RetriableStream.java:69) and SynchronizationContext.execute is executeLater(task); drain(),
the Runnable runs inline on the cancelling thread, and the uncaught handler at
RetriableStream.java:70-76 rethrows. The exception therefore surfaces on the caller's thread
rather than being contained.

The window is only reachable from application code because ClientCalls hands out the
ClientCallStreamObserver through ClientResponseObserver.beforeStart(), which runs in the
StreamObserverToCallListenerAdapter constructor (ClientCalls.java:451) — before startCall()
calls call.start() (ClientCalls.java:311). gRPC defers its own cancellation sources past
start() for exactly this reason, per the comment at ClientCallImpl.java:287:

Delay any sources of cancellation after start(), because most of the transports are broken if they
receive cancel before start. Issue #1343 has more details

but an application cancel arriving through the stub API gets none of that protection.

Only affects channels with retries enabled (the default), since RetriableStream is otherwise not
in the path.

Reproducer

Races the window and hits it within a few hundred attempts. No server is needed — the crash happens
before the transport is involved.

private object StringMarshaller : MethodDescriptor.Marshaller<String> {
    override fun stream(value: String): InputStream = ByteArrayInputStream(value.toByteArray())
    override fun parse(stream: InputStream): String = stream.readBytes().decodeToString()
}

private val METHOD: MethodDescriptor<String, String> = MethodDescriptor.newBuilder<String, String>()
    .setType(MethodDescriptor.MethodType.UNARY)
    .setFullMethodName("repro.CancelRace/Start")
    .setRequestMarshaller(StringMarshaller)
    .setResponseMarshaller(StringMarshaller)
    .build()

val channel = OkHttpChannelBuilder.forAddress("localhost", 1).usePlaintext().build()
val canceller = Executors.newSingleThreadExecutor()

repeat(20_000) { attempt ->
    val stream = AtomicReference<ClientCallStreamObserver<String>?>(null)
    val release = AtomicBoolean(false)

    val task = canceller.submit {
        while (!release.get()) Thread.onSpinWait()
        repeat(attempt % 64) { Thread.onSpinWait() }   // sweep the offset across attempts
        stream.get()?.cancel("cancel racing start", null)   // <-- throws here
    }

    val observer = object : ClientResponseObserver<String, String> {
        override fun beforeStart(requestStream: ClientCallStreamObserver<String>) {
            stream.set(requestStream)
            release.set(true)
        }
        override fun onNext(value: String) = Unit
        override fun onError(t: Throwable) = Unit
        override fun onCompleted() = Unit
    }

    try {
        ClientCalls.asyncUnaryCall(channel.newCall(METHOD, CallOptions.DEFAULT), "request", observer)
    } catch (_: IllegalStateException) {
        // cancel landed before start(): "call was cancelled". Not the bug.
    }
    task.get()
}

For a fully deterministic demonstration, a breakpoint at RetriableStream.java:388 with a
thread-only suspend policy holds the window open indefinitely; cancelling while it is parked
reproduces it every time.

Suggested direction

Either null-check masterListener in safeCloseMasterListener and fall back to
savedCloseMasterListenerReason once start() attaches the listener, or have
CallToStreamObserverAdapter.cancel() defer until the call has started, mirroring how
ClientCallImpl already defers its own cancellation sources past stream.start().