refactor(gax): remove circular ref between resumable upload future and chunk coordinator

[whowes]

This cleans up and helps clarify the layering and responsibility structure ahead of introducing non-happy path features.

In general the Future coordinates the overall upload lifecycle while delegating details of specific operations (e.g. chunk uploads, status listeners, global timeout) to the layer below. Actors on that layer don't maintain explicit references to the Future.

[whowes]

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors the resumable upload coordination by decoupling ResumableUploadChunkCoordinator from ResumableUploadFutureImpl. The coordinator now manages its own internal state and returns an ApiFuture upon starting, while ResumableUploadFutureImpl has been updated to observe this future and has removed the setInFlightFuture method. However, the review identifies two critical race conditions introduced by this refactoring: one in ResumableUploadChunkCoordinator where a cancellation of the result future after the initial check but before assigning currentChunkFuture can leak a background upload task, and another in ResumableUploadFutureImpl where a cancellation during coordinator.start() is not properly propagated to the newly created upload future. Both issues require adding explicit cancellation checks and propagation logic to prevent background resource leaks.

[gemini-code-assist]

There is a race condition where result can be cancelled after result.isDone() is checked at the beginning of transmitChunk, but before currentChunkFuture is assigned. In this scenario, the cancellation listener registered in start() will have already executed (finding currentChunkFuture to be null or a previous chunk), and the newly created chunkFuture will never be cancelled, leading to a leaked background upload task.

To prevent this, check if result has been cancelled immediately after assigning currentChunkFuture and cancel the chunk future if so.

      this.currentChunkFuture = chunkFuture;
      if (result.isCancelled()) {
        chunkFuture.cancel(true);
      }

[whowes]

Fixed.

[gemini-code-assist]

With the removal of setInFlightFuture, there is a race condition where resultFuture can be cancelled while coordinator.start() is executing. If this happens, resultFuture.isDone() will be true when entering the synchronized block, and the method will return early without cancelling the newly started uploadFuture. This can leak the upload process in the background.

To fix this, check if resultFuture was cancelled when it is done, and propagate the cancellation to uploadFuture accordingly.

            boolean shouldCancel = false;
            synchronized (lock) {
              if (resultFuture.isDone()) {
                shouldCancel = resultFuture.isCancelled();
              } else {
                inFlightFuture = uploadFuture;
              }
            }
            if (shouldCancel) {
              uploadFuture.cancel(true);
              return;
            }

References

1. When concurrent operations (such as cancellation and lazy initialization) are protected by a common lock (e.g., synchronized (this)), atomic state transitions (like compareAndSet) are not strictly necessary as the synchronization already prevents concurrent interleaving.

[whowes]

Fixed.

[blakeli0]

nit: rename result to something more readable, otherwise it's not easy to comprehend. There might be a more concise name for it, but IIUC, it's basically a uploadAllChunksResultFuture?

[whowes]

Renamed to uploadResultFuture.

[blakeli0]

Is the purpose of inFlightFuture to keep track of the in flight future so we know which future to cancel when cancel() is called?

In general, I feel we have a lot of complex logics just to support correct cancellation of this future. Maybe we can restructure the code to make it simpler.

[whowes]

Yes, your understanding of inFlightFuture is correct.

I simplified the handoff by exposing coordinator.getFuture() before calling coordinator.start(), so we swap inFlightFuture in a single synchronized block and eliminate the extra post-start cancellation check. I looked into a few other options (some transformAsync, some using listeners) but I feel like they led to denser, less-readable code overall, particularly when adding the features in later PRs.

[blakeli0]

I guess the only possibility that resultFuture is already done at this moment is that it is cancelled?

[whowes]

For this PR yes; once we add in the global timeout (#14425) this would also apply in that case.

[blakeli0]

Shall we wrap the whole transmitChunk method with try catch? Otherwise the uploadResultFuture may never complete if there are unhandled runtime exceptions outside of this try catch block.

[whowes]

Done.

[sonarqubecloud]

Quality Gate failed for 'gapic-generator-java-root'

Failed conditions
B Reliability Rating on New Code (required ≥ A)

See analysis details on SonarQube Cloud

Catch issues before they fail your Quality Gate with our IDE extension SonarQube for IDE

[sonarqubecloud]

Quality Gate failed for 'gapic-generator-java-root'

Failed conditions
0.0% Coverage on New Code (required ≥ 80%)
B Reliability Rating on New Code (required ≥ A)

See analysis details on SonarQube Cloud

Catch issues before they fail your Quality Gate with our IDE extension SonarQube for IDE