refactor: replace CleanupListener.closed flag with chain.isDone()

VOperationImpl.start needs to detect whether chain.start synchronously
delivered a terminal onClose, so it can skip registering the gRPC
cancellation listener (which would otherwise leak onto grpcContext).
Previously this was tracked via a 'closed' flag on CleanupListener — a
piggyback bookkeeping field on the listener wrapper that exists only
for VOperationImpl's coordination.

Add isDone() to the VRpc interface and ask the chain directly. The
chain is the natural source of truth for its own terminal state.
CleanupListener shrinks back to its single concern: relay events and
unhook the gRPC cancellation listener on close.

Implementations: RetryingVRpc delegates to currentState.isDone();
VRpcImpl reports state==CLOSED; ForwardingVRpc forwards; PendingVRpc
defers to realCall once handed off, otherwise reports isCancelled.
Test fakes (DelayedVRpc, FakeVRpc, anonymous VOperationImplTest chains)
implement the new method.

Drive-by: drop the defensive handling of tracer.onOperationStart throws
from RetryingVRpc.start, and the symmetric `!started` early-return in
Done.onStart that paired with it. CompositeVRpcTracer catches throws
from every child tracer, so the only way tracer.onOperationStart reaches
RetryingVRpc with a real throw is a test that bypasses Composite. Dead
code in production; relying on the existing chain.cancel cascade is
simpler than maintaining a separate short-circuit path.

Also: the already-started error in RetryingVRpc.start now dispatches
listener.onClose through ctx.getExecutor() rather than invoking it
synchronously on the caller, matching the dispatch convention used
everywhere else in the chain.

Author: igorbernstein2

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/ForwardingVRpc.java

@@ -36,6 +36,11 @@ public void cancel(@Nullable String message, @Nullable Throwable cause) {
     delegate.cancel(message, cause);
   }
 
+  @Override
+  public boolean isDone() {
+    return delegate.isDone();
+  }
+
   @Override
   public void requestNext() {
     delegate.requestNext();

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpc.java

@@ -69,9 +69,10 @@ public RetryingVRpc(Supplier<VRpc<ReqT, RespT>> supplier, BigtableTimer timer) {
   @Override
   public void start(ReqT req, VRpcCallContext ctx, VRpcListener<RespT> listener) {
     if (started) {
-      listener.onClose(
+      VRpcResult alreadyStarted =
           VRpcResult.createRejectedError(
-              Status.FAILED_PRECONDITION.withDescription("operation is already started")));
+              Status.FAILED_PRECONDITION.withDescription("operation is already started"));
+      ctx.getExecutor().execute(() -> listener.onClose(alreadyStarted));
       return;
     }
 
@@ -117,6 +118,11 @@ public void cancel(@Nullable String message, @Nullable Throwable cause) {
                 Status.CANCELLED.withDescription(message).withCause(finalCause))));
   }
 
+  @Override
+  public boolean isDone() {
+    return currentState.isDone();
+  }
+
   @Override
   public void requestNext() {
     // Assert the op-executor affinity even though the body is dead today — when streaming lands
@@ -389,10 +395,6 @@ class Done extends State {
 
     @Override
     public void onStart() {
-      if (!started) {
-        LOG.fine("operation is not started yet.");
-        return;
-      }
       // Per-attempt tracer pairing is owned by Active.onExit; Done just runs the user listener
       // and the per-operation tracer finish.
       Stopwatch appTimer = Stopwatch.createStarted();

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VOperationImpl.java

@@ -95,21 +95,21 @@ public void start(ReqT req, VRpcListener<RespT> listener) {
     // Register AFTER chain.start so a context-cancel that fires immediately is sequenced behind
     // start. Matches ClientCallImpl's ordering (grpc-java issue #1343).
     //
-    // Queueing the registration onto the op executor is what makes the closed-check sound: any
+    // Queueing the registration onto the op executor is what makes the isDone-check sound: any
     // onClose that chain.start enqueued during runInline drains FIRST (FIFO), so by the time this
-    // task runs wrapped.closed reflects whether onClose has already fired. If it has, we skip
-    // addListener — otherwise the listener would pin the chain on grpcContext for its lifetime
-    // (CleanupListener.onClose already called removeListener as a no-op pre-registration). If
-    // grpcContext gets cancelled between start() returning and this task running, the
-    // directExecutor below fires the listener immediately on addListener, so cancel still
-    // propagates correctly.
+    // task runs chain.isDone() reflects whether the chain has already reached terminal. If it
+    // has, we skip addListener — otherwise the listener would pin the chain on grpcContext for
+    // its lifetime (CleanupListener.onClose already called removeListener as a no-op
+    // pre-registration). If grpcContext gets cancelled between start() returning and this task
+    // running, the directExecutor below fires the listener immediately on addListener, so cancel
+    // still propagates correctly.
     //
     // runInline is the right verb here: when chain.start enqueued nothing (common path), the
     // executor is idle and the body runs inline on this thread — no extra context switch. When
     // chain.start did enqueue an onClose, runInline takes the queue branch and FIFO drains both.
     exec.runInline(
         () -> {
-          if (!wrapped.closed) {
+          if (!chain.isDone()) {
             grpcContext.addListener(cancellationListener, MoreExecutors.directExecutor());
           }
         });
@@ -123,10 +123,6 @@ public void cancel(@Nullable String message, @Nullable Throwable cause) {
   private static class CleanupListener<RespT> extends ForwardListener<RespT> {
     private final Context grpcContext;
     private final Context.CancellationListener cancellationListener;
-    // Read by VOperationImpl.start on the caller thread after runInline returns. runInline runs
-    // chain.start synchronously, so any sync onClose has completed (and this flag been set) by
-    // the time start() reads it on the same thread — no synchronization needed.
-    volatile boolean closed = false;
 
     CleanupListener(
         VRpcListener<RespT> delegate,
@@ -139,7 +135,6 @@ private static class CleanupListener<RespT> extends ForwardListener<RespT> {
 
     @Override
     public void onClose(VRpcResult result) {
-      closed = true;
       grpcContext.removeListener(cancellationListener);
       super.onClose(result);
     }

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java

@@ -59,6 +59,14 @@ public interface VRpc<ReqT, RespT> {
   /** Cancel a started RPC. This will be done by best effort. */
   void cancel(@Nullable String message, @Nullable Throwable cause);
 
+  /**
+   * True once a terminal result has been (or is about to be) delivered to the listener; future
+   * events on this VRpc are no-ops. Callers use this to detect a synchronous terminal during
+   * {@link #start} — e.g. VOperationImpl checks this to avoid registering a gRPC cancellation
+   * listener that would never be removed because the chain already finished.
+   */
+  boolean isDone();
+
   /**
    * TBD - server streaming rpcs. This will be used to request more data. Unlike gRPC's request(n),
    * starting a call will implicitly request the first message.

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImpl.java

@@ -775,6 +775,13 @@ void cancelWithResult(VRpcResult result) {
               });
     }
 
+    @Override
+    public boolean isDone() {
+      // realCall set in drainTo's lambda; once we hand off, it's the source of truth.
+      // Pre-handoff, isCancelled tracks our own terminal state.
+      return realCall != null ? realCall.isDone() : isCancelled;
+    }
+
     @Override
     public void requestNext() {
       if (realCall != null) {

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/session/VRpcImpl.java

@@ -196,6 +196,11 @@ public void cancel(@Nullable String message, @Nullable Throwable cause) {
     session.cancelRpc(rpcId, message, cause);
   }
 
+  @Override
+  public boolean isDone() {
+    return state.get() == State.CLOSED;
+  }
+
   @Override
   public void requestNext() {
     throw new UnsupportedOperationException("streamed RPCs are not supported yet");

java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/api/TableBaseTest.java

@@ -219,6 +219,11 @@ public void start(Object req, VRpcCallContext ctx, VRpcListener ignored) {
     @Override
     public void cancel(@Nullable String message, @Nullable Throwable cause) {}
 
+    @Override
+    public boolean isDone() {
+      return false;
+    }
+
     @Override
     public void requestNext() {}
   }

java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/VRpcTracerTest.java

@@ -510,6 +510,11 @@ public void cancel(@Nullable String message, @Nullable Throwable cause) {
       throw new UnsupportedOperationException();
     }
 
+    @Override
+    public boolean isDone() {
+      return false;
+    }
+
     @Override
     public void requestNext() {
       throw new UnsupportedOperationException();

java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/middleware/VOperationImplTest.java

@@ -57,6 +57,11 @@ public void cancel(@Nullable String msg, @Nullable Throwable cause) {
             cancelLatch.countDown();
           }
 
+          @Override
+          public boolean isDone() {
+            return false;
+          }
+
           @Override
           public void requestNext() {}
         };
@@ -88,9 +93,9 @@ public void onClose(VRpcResult result) {}
 
   @Test
   void asyncOnCloseFromChainDoesNotPropagateLaterContextCancel() throws InterruptedException {
-    // Regression for the wrapped.closed TOCTOU. When chain.start asynchronously queues an
+    // Regression for the chain-already-done TOCTOU. When chain.start asynchronously queues an
     // onClose via the op executor, the addListener task (also queued through the op executor
-    // by VOperationImpl.start) drains FIFO after the onClose and observes wrapped.closed=true,
+    // by VOperationImpl.start) drains FIFO after the onClose and observes chain.isDone()=true,
     // so the cancellationListener is NOT registered. A later grpcContext.cancel therefore has
     // no path to reach chain.cancel — which is correct because the chain has already terminated.
     Context.CancellableContext grpcContext = Context.current().withCancellation();
@@ -100,22 +105,31 @@ void asyncOnCloseFromChainDoesNotPropagateLaterContextCancel() throws Interrupte
 
     VRpc<String, String> chain =
         new VRpc<String, String>() {
+          private volatile boolean done = false;
+
           @Override
           public void start(String req, VRpcCallContext ctx, VRpcListener<String> listener) {
             // Simulate PendingVRpc pool-closed branch / VRpcImpl deadline short-circuit.
             ctx.getExecutor()
                 .execute(
-                    () ->
-                        listener.onClose(
-                            VRpcResult.createUncommitedError(
-                                Status.UNAVAILABLE.withDescription("fast-fail"))));
+                    () -> {
+                      done = true;
+                      listener.onClose(
+                          VRpcResult.createUncommitedError(
+                              Status.UNAVAILABLE.withDescription("fast-fail")));
+                    });
           }
 
           @Override
           public void cancel(@Nullable String msg, @Nullable Throwable cause) {
             chainCancelCount.incrementAndGet();
           }
 
+          @Override
+          public boolean isDone() {
+            return done;
+          }
+
           @Override
           public void requestNext() {}
         };