Rollback a reactive transaction on pipeline cancel

... instead of committing. This fixes atomicity violation problems. See spring-projects#25091 and https://stackoverflow.com/questions/61822249/spring-reactive-transaction-gets-committed-on-cancel-producing-partial-commits?noredirect=1#comment109381171_61822249
rpuch · May 18, 2020 · 95c2872 · 95c2872
1 parent 0ddc62e
commit 95c2872
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 3 deletions.
diff --git a/...x/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java b/...x/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java
@@ -876,7 +876,7 @@ public Object invokeWithinTransaction(Method method, @Nullable Class<?> targetCl
 										},
 										this::commitTransactionAfterReturning,
 										(txInfo, err) -> Mono.empty(),
-										this::commitTransactionAfterReturning)
+										this::rollbackTransactionDueToCancel)
 										.onErrorResume(ex ->
 												completeTransactionAfterThrowing(it, ex).then(Mono.error(ex)));
 							}
@@ -906,7 +906,7 @@ public Object invokeWithinTransaction(Method method, @Nullable Class<?> targetCl
 											},
 											this::commitTransactionAfterReturning,
 											(txInfo, ex) -> Mono.empty(),
-											this::commitTransactionAfterReturning)
+											this::rollbackTransactionDueToCancel)
 									.onErrorResume(ex ->
 											completeTransactionAfterThrowing(it, ex).then(Mono.error(ex)));
 						}
@@ -973,6 +973,16 @@ private Mono<Void> commitTransactionAfterReturning(@Nullable ReactiveTransaction
 			return Mono.empty();
 		}
 
+		private Mono<Void> rollbackTransactionDueToCancel(@Nullable ReactiveTransactionInfo txInfo) {
+			if (txInfo != null && txInfo.getReactiveTransaction() != null) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("Rolling transaction back for [" + txInfo.getJoinpointIdentification() + "] due to cancel");
+				}
+				return txInfo.getTransactionManager().rollback(txInfo.getReactiveTransaction());
+			}
+			return Mono.empty();
+		}
+
 		private Mono<Void> completeTransactionAfterThrowing(@Nullable ReactiveTransactionInfo txInfo, Throwable ex) {
 			if (txInfo != null && txInfo.getReactiveTransaction() != null) {
 				if (logger.isTraceEnabled()) {

diff --git a/...-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java b/...-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java
@@ -79,7 +79,7 @@ public <T> Mono<T> transactional(Mono<T> mono) {
 			// Need re-wrapping of ReactiveTransaction until we get hold of the exception
 			// through usingWhen.
 			return status.flatMap(it -> Mono.usingWhen(Mono.just(it), ignore -> mono,
-					this.transactionManager::commit, (res, err) -> Mono.empty(), this.transactionManager::commit)
+					this.transactionManager::commit, (res, err) -> Mono.empty(), this.transactionManager::rollback)
 					.onErrorResume(ex -> rollbackOnException(it, ex).then(Mono.error(ex))));
 		})
 		.subscriberContext(TransactionContextManager.getOrCreateContext())