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

### 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:

```java
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.

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

private val METHOD: MethodDescriptor = MethodDescriptor.newBuilder()
.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?>(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 {
override fun beforeStart(requestStream: ClientCallStreamObserver) {
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()`.

まず ClientCallImpl.startInternal と RetriableStream.start、cancel、safeCloseMasterListener を読んでから、リトライを有効にして提供された Kotlin reproducer を実行します。提案された 2 つの方向性を比較し、ClientCall.start と競合するキャンセルによって NPE が発生しなくなり、呼び出しがキャンセルされるか安全に無視されることを確認します。

索引モデルが issue の本文から書いたものです。