Improve transaction failure handling

If Bolt exchange fails for whatever reason, subsequent runs should emit a failed transaction error.

Author: injectives

driver/src/main/java/org/neo4j/driver/internal/async/TerminationAwareResponseHandler.java

@@ -28,6 +28,7 @@ final class TerminationAwareResponseHandler extends DelegatingResponseHandler {
     private final Logger log;
     private final TerminationAwareStateLockingExecutor executor;
     private final Consumer<Throwable> throwableConsumer;
+    private boolean hadError;
 
     TerminationAwareResponseHandler(
             Logging logging,
@@ -42,6 +43,7 @@ final class TerminationAwareResponseHandler extends DelegatingResponseHandler {
 
     @Override
     public void onError(Throwable throwable) {
+        hadError = true;
         throwableConsumer.accept(Futures.completionExceptionCause(throwable));
         super.onError(throwable);
     }
@@ -50,10 +52,16 @@ public void onError(Throwable throwable) {
     public void onComplete() {
         var throwable = executor.execute(Function.identity());
         if (throwable != null) {
-            log.trace(
-                    "Reporting an existing %s error to delegate",
-                    throwable.getClass().getCanonicalName());
-            delegate.onError(throwable);
+            if (hadError) {
+                log.trace(
+                        "Skipping reporting %s error because an actual error has already been reported",
+                        throwable.getClass().getCanonicalName());
+            } else {
+                log.trace(
+                        "Reporting an existing %s error to delegate",
+                        throwable.getClass().getCanonicalName());
+                delegate.onError(throwable);
+            }
         }
         log.trace("Completing delegate");
         delegate.onComplete();

driver/src/main/java/org/neo4j/driver/internal/async/UnmanagedTransaction.java

@@ -320,7 +320,10 @@ public CompletionStage<Void> terminateAsync() {
 
     @Override
     public <T> T execute(Function<Throwable, T> causeOfTerminationConsumer) {
-        return executeWithLock(lock, () -> causeOfTerminationConsumer.apply(causeOfTermination));
+        return executeWithLock(lock, () -> {
+            var throwable = causeOfTermination == null ? null : failedTxException(causeOfTermination);
+            return causeOfTerminationConsumer.apply(throwable);
+        });
     }
 
     private void ensureCanRunQueries() {
@@ -347,15 +350,7 @@ private void ensureCanRunQueries() {
                 if (causeOfTermination instanceof TransactionTerminatedException transactionTerminatedException) {
                     throw transactionTerminatedException;
                 } else {
-                    var message =
-                            "Cannot run more queries in this transaction, it has either experienced an fatal error or was explicitly terminated";
-                    throw new TransactionTerminatedException(
-                            GqlStatusError.UNKNOWN.getStatus(),
-                            GqlStatusError.UNKNOWN.getStatusDescription(message),
-                            "N/A",
-                            message,
-                            GqlStatusError.DIAGNOSTIC_RECORD,
-                            causeOfTermination);
+                    throw failedTxException(causeOfTermination);
                 }
             } else if (commitFuture != null) {
                 var message = "Cannot run more queries in this transaction, it is being committed";
@@ -578,6 +573,18 @@ private CompletionStage<Void> closeAsync(boolean commit, boolean completeWithNul
         return stage;
     }
 
+    private static TransactionTerminatedException failedTxException(Throwable cause) {
+        var message =
+                "Cannot run more queries in this transaction, it has either experienced a fatal error or was explicitly terminated";
+        return new TransactionTerminatedException(
+                GqlStatusError.UNKNOWN.getStatus(),
+                GqlStatusError.UNKNOWN.getStatusDescription(message),
+                "N/A",
+                message,
+                GqlStatusError.DIAGNOSTIC_RECORD,
+                cause);
+    }
+
     private static class BeginResponseHandler implements DriverResponseHandler {
         final CompletableFuture<UnmanagedTransaction> summaryFuture = new CompletableFuture<>();
         private final ApiTelemetryWork apiTelemetryWork;

driver/src/test/java/org/neo4j/driver/integration/TransactionIT.java

@@ -422,7 +422,7 @@ void shouldPreventPullAfterTransactionTermination(boolean iterate) {
                     result.list();
                 }
             });
-            assertEquals(terminationException, exception);
+            assertEquals(terminationException, exception.getCause());
         }
         tx.close();
     }
@@ -441,8 +441,8 @@ void shouldPreventDiscardAfterTransactionTermination() {
 
         // Then
         for (var result : List.of(result0, result1)) {
-            var exception = assertThrows(ClientException.class, result::consume);
-            assertEquals(terminationException, exception);
+            var exception = assertThrows(TransactionTerminatedException.class, result::consume);
+            assertEquals(terminationException, exception.getCause());
         }
         tx.close();
     }

driver/src/test/java/org/neo4j/driver/integration/reactive/ReactiveTransactionIT.java

@@ -63,7 +63,7 @@ void shouldPreventPullAfterTransactionTermination() {
         for (var result : List.of(result0, result1)) {
             var exception = assertThrows(
                     ClientException.class, () -> Flux.from(result.records()).blockFirst());
-            assertEquals(terminationException, exception);
+            assertEquals(terminationException, exception.getCause());
         }
         Mono.fromDirect(tx.close()).block();
     }
@@ -92,7 +92,7 @@ void shouldPreventDiscardAfterTransactionTermination() {
         for (var result : List.of(result0, result1)) {
             var exception = assertThrows(ClientException.class, () -> Mono.fromDirect(result.consume())
                     .block());
-            assertEquals(terminationException, exception);
+            assertEquals(terminationException, exception.getCause());
         }
         Mono.fromDirect(tx.close()).block();
     }