Fix residual leak in downloadDirectory by skipping file downloads after cancellation.

[RanVaknin]

In #6875, we addressed canceling in flight transfers when a directory download is cancelled. That fix reduced the leaked files per orphaned directory from thousands to single digits in stress test. However there was a residual leak for downloads that were already dispatched to doDownloadSingleFile (the method that creates the destination directory and initiates the file download) before the cancellation signal propagated.

When cancellation occurs, thread A calls subscription.cancel() to stop new items from being delivered. Meanwhile, thread B has already picked up an item from onNext and is inside doDownloadSingleFile. Since it entered before thread A's cancellation took effect, it creates the destination directory and starts a download into it.

This PR adds an isCompletedExceptionally() guard at the top of doDownloadSingleFile so that any thread that entered the method before cancellation took effect will return early before touching the filesystem.

The PR also adds a isCompletedExceptionally() check in AsyncBufferingSubscriber.onNext, immediately after adding the future to requestsInFlight. When returnFuture completes exceptionally, the cancel handler iterates requestsInFlight and cancels every future in it. That iteration only happens once. If it already ran before the future was added to the set, the future was missed. The additional check ensures we cancel it in that case.

The tests:

1. Updated downloadDirectory_cancel_shouldCancelAllFutures - now waits for both downloads to start before cancelling. The original test called cancel() immediately after downloadDirectory(), implicitly assuming all downloads would always start regardless of cancellation state. With the new isCompletedExceptionally() guard, that assumption is not always true since cancel can now prevent downloads from starting. The test's intent is unchanged (in flight futures get cancelled), we just make sure both futures are actually in flight before cancelling. (Sanity check tested this with a repeated test suite 1000 times to make sure its deterministic)

2. Added downloadDirectory_cancelledFuture_shouldNotCreateDirectories - The test cancels the directory download future, then delivers an S3 object through the publisher. Asserts that the destination subdirectory was never created on the filesystem. Without the fix, doDownloadSingleFile would call createParentDirectoriesIfNeeded() and create the directory despite the cancellation.

3. Added returnFutureCancelledDuringOnNext_shouldCancelInFlightFuture in AsyncBufferingSubscriberTest - Forces the race outcome by triggering cancellation inside consumer.apply(), guaranteeing the cancel handler iterates an empty set. Asserts the future gets cancelled via the post add check.

[github-actions]

This pull request has been closed and the conversation has been locked. Comments on closed PRs are hard for our team to see. If you need more assistance, please open a new issue that references this one.