A similar default implementation existed in the past but using .next() instead of .singleOrEmpty(). This was changed in gh-23562 because .next() triggers the cancellation of the source.

I believe we could use .singleOrEmpty() which IIUC should not cancel the source before receiving the completion signal from the Mono.

Please let me know if this exceeds the scope of the issue and I would drop it from the PR.

I was hesitant about filing a PR because I wasn't sure I was fully understanding this part.

IIUC we could now simplify this because the rollbackOnException has been moved to inside the usingWhen.

Relates to this comment: https://github.com/spring-projects/spring-framework/pull/23562/files#r319925489

The previous code commits the transaction depending on successful/exceptional completion/cancellation. Any failure in the async cleanup methods (specifically commit because rollback happens upon cancellation) dropped into rollbackOnException which performs another round of cleanup and that is the actual bug.

To solve the issue, we need to call rollbackOnException in the transaction cleanup instead of onErrorResume(…) and (tx, ex) -> Mono.empty(). That would align the flow with the imperative TransactionTemplate.

It is possible that we need to refine the actual exception if it is emitted by the async clean up afterward.

Note that the same issue exists in TransactionAspectSupport.ReactiveTransactionSupport.

Thanks for the feedback @mp911de :) Do you mean something like this?

Flux.usingWhen(
	this.transactionManager.getReactiveTransaction(this.transactionDefinition),
	action::doInTransaction,
	this.transactionManager::commit,
	this::rollbackOnException,
	this.transactionManager::rollback))

Was indeed going for something like this initially but as you mentioned it would require refining the exception, because a failure on either asyncComplete or asyncError wraps the exception like this:

java.lang.RuntimeException: Async resource cleanup failed after onComplete|onError

Maybe that's ok? But that's why I thought that the .onErrorResume(rollback) + .concatWith(commit) combination may achieve something similar and avoid the exception refining. Do you think it would be preferable to go with the asyncComplete&asyncError + exception cleanup approach?

Note that the same issue exists in TransactionAspectSupport.ReactiveTransactionSupport.

Good point! Shall I apply these changes also over there, or do you think separate PR?

That bit looks decent. The more operators we use the more performance impact we can generate. Regarding exception mapping, I'm not sure which top-level exception should be propagated downstream. Looking into the transaction manager, a RuntimeException seems fine. Maybe @jhoeller can provide a bit more guidance.

Shall I apply these changes also over there, or do you think separate PR?

The issue is the same and it makes sense to fix the same problem in multiple places at once as we can keep the context within one commit from your side.

Pushed the changes. Went ahead with a message-based approach to the unwrapping, maybe there is a utility for this? 🤔 Didn't find anything in reactor.core.Exceptions.

Paging @simonbasle, maybe Simon can give further insights.

there isn't such an unwrapping utility currently, not for the sake of detecting that particular kind of exception.

@EnricSala note that in the case of a rollback which fails, we have a RuntimeException with:

1. the cause of the rollback failure as the getCause()
2. the original exception in the usingWhen which triggered the rollback in the first place (the rollbackCause) as a suppressed exception

Thanks for the reply! Agree, the exception propagated by usingWhen meets the criteria so leaving it untouched would be an option 👍

I squashed the changes on the branch leaving only two commits: the first one shows the implementation with exception unwrapping and the second commit shows what it would look like if we drop the unwrapping.

Both options resolve my problem, so I think at this point it may be a matter of preference or consistency with the rest of the framework. Please let me know which option would be preferable :)

It makes sense to unwrap the exception as callers typically do not expect a generic RuntimeException.

I would also have a preference for this option because the RuntimeException feels a bit synthetic, it's only there due to the usingWhen implementation, and it doesn't align with the behavior of non-reactive transactions.

I have applied the change, it should be ready for review :)