Fix directoryDownload cancel leak

[RanVaknin]

When a DirectoryDownload (or DirectoryUpload) future is cancelled or completed exceptionally, AsyncBufferingSubscriber cancels in flight file transfer futures but does not cancel the upstream Subscription.

This means the listObjectsV2 paginator continues delivering objects, and new file downloads continue to be started after the user has cancelled the operation.

This change adds subscription.cancel() to the whenComplete handler so that cancellation stops the entire pipeline of both inflight transfers and new work from the upstream publisher.

Conforms to Reactive Streams spec:

6 A Subscriber MUST call Subscription.cancel() if the Subscription is no longer needed.

[RanVaknin]

subscription.cancel() now exists in two codepaths:

1. The existing codepath within onNext() catch block.
2. The new addition in future.whenComplete()

Flow:

- onNext()
-- catch
-- subscription.cancel()
-- onError()

- onError()
-- completes exceptionally
-- future.whenCompleted()
-- subscription.cancel() // <-- new addition

I believe this is safe to call twice because:

5 Subscription.cancel MUST respect the responsivity of its caller by returning in a timely manner, MUST be idempotent and MUST be thread-safe.

Note: the codepath which the new addition solves is when the cancellation signal comes from outside of the pub/sub pipeline. If a user aborts the operation by cancelling the DirectoryDownload completion future directly, neither onNext nor onError are involved. The whenComplete handler is the only place that reacts to this external cancellation.

[sonarqubecloud]

Quality Gate failed

Failed conditions
34 Security Hotspots
C Security 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

[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.