Improve handling of Errors

driver-benchmarks/src/main/com/mongodb/benchmark/framework/MongoCryptBenchmarkRunner.java

@@ -67,8 +67,8 @@ private static String getFileAsString(final String fileName) {
                 throw new RuntimeException("Could not find file " + fileName);
             }
             return new String(Files.readAllBytes(Paths.get(resource.toURI())));
-        } catch (Throwable t) {
-            throw new RuntimeException("Could not parse file " + fileName, t);
+        } catch (Exception e) {
+            throw new RuntimeException("Could not parse file " + fileName, e);
         }
     }
 

driver-core/src/main/com/mongodb/internal/ExceptionUtils.java

@@ -26,6 +26,8 @@
 import org.bson.BsonString;
 import org.bson.BsonValue;
 
+import java.io.PrintStream;
+import java.lang.Thread.UncaughtExceptionHandler;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -46,6 +48,42 @@ public static boolean isOperationTimeoutFromSocketException(final Throwable e) {
         return e instanceof MongoOperationTimeoutException && e.getCause() instanceof MongoSocketException;
     }
 
+    /**
+     * We must always propagate {@link Error}s, instead of wrapping them in / replacing them with {@link Exception}s, or swallowing them.
+     * <p>
+     * This enables an application to decide how to deal with {@link Error} via {@link UncaughtExceptionHandler}.
+     * An {@link Error} is more likely to cause an invariant violation than an {@link Exception},
+     * because it is less likely to be taken into account in code. An {@link AssertionError} outright informs about an invariant violation.
+     * Furthermore, a {@link VirtualMachineError} not only may happen in a peculiar situation,
+     * but also may be <a href="https://docs.oracle.com/javase/specs/jls/se17/html/jls-11.html#jls-11.1.3">asynchronous</a>.
+     * That is why it may be a good idea for an application to terminate on {@link Error}.
+     * We cannot make such a decision for an application, but we must do our best to give it an opportunity to react to an {@link Error}.
+     * <p>
+     * If there is no {@link UncaughtExceptionHandler}, then the uncaught exception is
+     * {@linkplain Throwable#printStackTrace(PrintStream) printed}
+     * to {@link System#err}, see {@link ThreadGroup#uncaughtException(Thread, Throwable)}.
+     *
+     * @throws Error Iff {@code t} is {@link Error}.
+     * @see #mapUnlessError(Throwable, Function)
+     */
+    public static void rethrowIfError(final Throwable t) {
+        if (t instanceof Error) {
+            throw (Error) t;
+        }
+    }
+
+    /**
+     * See {@link #rethrowIfError(Throwable)}.
+     *
+     * @param mapper Is used iff {@code t} is not {@link Error}.
+     */
+    public static Throwable mapUnlessError(final Throwable t, final Function<Throwable, ? extends Exception> mapper) {
+        if (t instanceof Error) {
+            return t;
+        }
+        return mapper.apply(t);
+    }
+
     public static final class MongoCommandExceptionUtils {
         public static int extractErrorCode(final BsonDocument response) {
             return extractErrorCodeAsBson(response).intValue();

driver-core/src/main/com/mongodb/internal/async/ErrorHandlingResultCallback.java

@@ -20,6 +20,7 @@
 import com.mongodb.lang.Nullable;
 
 import static com.mongodb.assertions.Assertions.notNull;
+import static com.mongodb.internal.ExceptionUtils.rethrowIfError;
 
 /**
  * <p>This class is not part of the public API and may be removed or changed at any time.</p>
@@ -47,6 +48,7 @@ public void onResult(@Nullable final T result, @Nullable final Throwable t) {
             wrapped.onResult(result, t);
         } catch (Throwable e) {
             logger.error("Callback onResult call produced an error", e);
+            rethrowIfError(e);
         }
     }
 

driver-core/src/main/com/mongodb/internal/capi/MongoCryptHelper.java

@@ -199,8 +199,8 @@ public static void startProcess(final ProcessBuilder processBuilder) {
             processBuilder.redirectErrorStream(true);
             processBuilder.redirectOutput(new File(System.getProperty("os.name").startsWith("Windows") ? "NUL" : "/dev/null"));
             processBuilder.start();
-        } catch (Throwable t) {
-            throw new MongoClientException("Exception starting mongocryptd process. Is `mongocryptd` on the system path?", t);
+        } catch (Exception e) {
+            throw new MongoClientException("Exception starting mongocryptd process. Is `mongocryptd` on the system path?", e);
         }
     }
 

driver-core/src/main/com/mongodb/internal/connection/BaseCluster.java

@@ -66,6 +66,7 @@
 import static com.mongodb.connection.ServerDescription.MAX_DRIVER_WIRE_VERSION;
 import static com.mongodb.connection.ServerDescription.MIN_DRIVER_SERVER_VERSION;
 import static com.mongodb.connection.ServerDescription.MIN_DRIVER_WIRE_VERSION;
+import static com.mongodb.internal.ExceptionUtils.rethrowIfError;
 import static com.mongodb.internal.Locks.withInterruptibleLock;
 import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE;
 import static com.mongodb.internal.connection.EventHelper.wouldDescriptionsGenerateEquivalentEvents;
@@ -185,6 +186,7 @@ public void selectServerAsync(final ServerSelector serverSelector, final Operati
             final SingleResultCallback<ServerTuple> callback) {
         if (isClosed()) {
             callback.onResult(null, new MongoClientException("Cluster was closed during server selection."));
+            return;
         }
 
         Timeout computedServerSelectionTimeout = operationContext.getTimeoutContext().computeServerSelectionTimeout();
@@ -280,7 +282,7 @@ private Timeout startMinWaitHeartbeatTimeout() {
     private boolean handleServerSelectionRequest(
             final ServerSelectionRequest request, final CountDownLatch currentPhase,
             final ClusterDescription description) {
-
+        boolean requestOnResultCalled = false;
         try {
             OperationContext operationContext = request.getOperationContext();
             if (currentPhase != request.phase) {
@@ -303,6 +305,7 @@ private boolean handleServerSelectionRequest(
                     ServerAddress serverAddress = serverTuple.getServerDescription().getAddress();
                     logServerSelectionSucceeded(operationContext, clusterId, serverAddress, request.originalSelector, description);
                     serverDeprioritization.updateCandidate(serverAddress);
+                    requestOnResultCalled = true;
                     request.onResult(serverTuple, null);
                     return true;
                 }
@@ -315,7 +318,10 @@ private boolean handleServerSelectionRequest(
                 logAndThrowTimeoutException(operationContext, request.originalSelector, description);
             });
             return false;
-        } catch (Exception e) {
+        } catch (Throwable e) {
+            if (requestOnResultCalled) {
+                throw e;
+            }
             request.onResult(null, e);
             return true;
         }
@@ -455,8 +461,9 @@ private static final class ServerSelectionRequest {
         void onResult(@Nullable final ServerTuple serverTuple, @Nullable final Throwable t) {
             try {
                 callback.onResult(serverTuple, t);
-            } catch (Throwable tr) {
-                // ignore
+            } catch (Throwable e) {
+                LOGGER.error("Callback onResult call produced an error", e);
+                rethrowIfError(e);
             }
         }
 

driver-core/src/main/com/mongodb/internal/connection/InternalStreamConnection.java

@@ -79,6 +79,8 @@
 import static com.mongodb.assertions.Assertions.assertNull;
 import static com.mongodb.assertions.Assertions.isTrue;
 import static com.mongodb.assertions.Assertions.notNull;
+import static com.mongodb.internal.ExceptionUtils.mapUnlessError;
+import static com.mongodb.internal.ExceptionUtils.rethrowIfError;
 import static com.mongodb.internal.TimeoutContext.createMongoTimeoutException;
 import static com.mongodb.internal.async.AsyncRunnable.beginAsync;
 import static com.mongodb.internal.async.ErrorHandlingResultCallback.errorHandlingCallback;
@@ -242,11 +244,8 @@ public void open(final OperationContext originalOperationContext) {
             initAfterHandshakeFinish(initializationDescription);
         } catch (Throwable t) {
             close();
-            if (t instanceof MongoException) {
-                throw (MongoException) t;
-            } else {
-                throw new MongoException(t.toString(), t);
-            }
+            rethrowIfError(t);
+            throw MongoException.fromThrowableNonNull(t);
         }
     }
 
@@ -961,12 +960,12 @@ public void completed(@Nullable final ByteBuf buffer) {
                 @Override
                 public void failed(final Throwable t) {
                     close();
-                    callback.onResult(null, translateReadFailure(t, operationContext));
+                    callback.onResult(null, mapUnlessError(t, e -> translateReadException(e, operationContext)));
                 }
             });
         } catch (Throwable t) {
             close();
-            callback.onResult(null, translateReadFailure(t, operationContext));
+            callback.onResult(null, mapUnlessError(t, e -> translateReadException(e, operationContext)));
         }
     }
 
@@ -990,17 +989,6 @@ private void updateSessionContext(final SessionContext sessionContext, final Res
         }
     }
 
-    /**
-     * Rethrows a fatal JVM {@link Error} (e.g. {@link OutOfMemoryError}) unchanged, so it is never downgraded to a
-     * catchable {@link MongoException}. Used by the paths that propagate a failure by throwing; the async read path
-     * delivers the failure as a callback value instead and uses {@link #translateReadFailure} for the same purpose.
-     */
-    private static void rethrowIfError(final Throwable t) {
-        if (t instanceof Error) {
-            throw (Error) t;
-        }
-    }
-
     private void throwTranslatedWriteException(final Throwable e, final OperationContext operationContext) {
         rethrowIfError(e);
         if (operationContext.getTimeoutContext().hasTimeoutMS()) {
@@ -1026,18 +1014,6 @@ private void throwTranslatedWriteException(final Throwable e, final OperationCon
         }
     }
 
-    /**
-     * Translates a read failure for delivery to an async callback. {@link Error}s are passed through unchanged
-     * rather than wrapped in a {@link MongoException}, so a fatal JVM error (e.g. {@link OutOfMemoryError}) is not
-     * downgraded to a catchable exception. The sync read path uses {@link #rethrowIfError} for the same purpose.
-     */
-    private Throwable translateReadFailure(final Throwable e, final OperationContext operationContext) {
-        if (e instanceof Error) {
-            return e;
-        }
-        return translateReadException(e, operationContext);
-    }
-
     private MongoSocketWriteTimeoutException createWriteTimeoutException(final SocketTimeoutException e) {
         return new MongoSocketWriteTimeoutException("Timeout while sending message", getServerAddress(), e);
     }

driver-core/src/main/com/mongodb/internal/connection/LoadBalancedCluster.java

@@ -61,6 +61,7 @@
 import static com.mongodb.assertions.Assertions.isTrue;
 import static com.mongodb.assertions.Assertions.notNull;
 import static com.mongodb.connection.ServerConnectionState.CONNECTING;
+import static com.mongodb.internal.ExceptionUtils.rethrowIfError;
 import static com.mongodb.internal.connection.BaseCluster.logServerSelectionStarted;
 import static com.mongodb.internal.connection.BaseCluster.logServerSelectionSucceeded;
 import static com.mongodb.internal.connection.BaseCluster.logTopologyMonitoringStopping;
@@ -445,16 +446,18 @@ OperationContext getOperationContext() {
         public void onSuccess(final ServerTuple serverTuple) {
             try {
                 callback.onResult(serverTuple, null);
-            } catch (Exception e) {
-                LOGGER.warn("Unanticipated exception thrown from callback", e);
+            } catch (Throwable t) {
+                LOGGER.error("Callback onResult call produced an error", t);
+                rethrowIfError(t);
             }
         }
 
         public void onError(final Throwable exception) {
             try {
                 callback.onResult(null, exception);
-            } catch (Exception e) {
-                LOGGER.warn("Unanticipated exception thrown from callback", e);
+            } catch (Throwable t) {
+                LOGGER.error("Callback onResult call produced an error", t);
+                rethrowIfError(t);
             }
         }
     }

driver-core/src/main/com/mongodb/internal/connection/tlschannel/async/AsynchronousTlsChannelGroup.java

@@ -58,6 +58,7 @@
 import java.util.function.Consumer;
 import java.util.function.LongConsumer;
 
+import static com.mongodb.internal.ExceptionUtils.rethrowIfError;
 import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException;
 import static java.lang.String.format;
 
@@ -233,6 +234,7 @@ public void execute(final Runnable r) {
                 r.run();
             } catch (Throwable t) {
                 LOGGER.error(null, t);
+                rethrowIfError(t);
             }
         });
     }
@@ -427,8 +429,9 @@ private void loop() {
                 processPendingInterests();
                 checkClosings();
             }
-        } catch (Throwable e) {
-            LOGGER.error("error in selector loop", e);
+        } catch (Throwable t) {
+            LOGGER.error(this + " stopped working. You may want to recreate the MongoClient", t);
+            rethrowIfError(t);
         } finally {
             executor.shutdown();
             // use shutdownNow to stop delayed tasks
@@ -467,6 +470,7 @@ private void processWrite(RegisteredSocket socket) {
                                 doWrite(socket, op);
                             } catch (Throwable e) {
                                 LOGGER.error("error in operation", e);
+                                rethrowIfError(e);
                             }
                         });
             }
@@ -486,6 +490,7 @@ private void processRead(RegisteredSocket socket) {
                                 doRead(socket, op);
                             } catch (Throwable e) {
                                 LOGGER.error("error in operation", e);
+                                rethrowIfError(e);
                             }
                         });
             }

driver-core/src/main/com/mongodb/internal/operation/ChangeStreamBatchCursor.java

@@ -40,6 +40,7 @@
 import java.util.function.Consumer;
 
 import static com.mongodb.assertions.Assertions.assertNotNull;
+import static com.mongodb.internal.ExceptionUtils.rethrowIfError;
 import static com.mongodb.internal.operation.ChangeStreamBatchCursorHelper.isResumableError;
 import static com.mongodb.internal.operation.SyncOperationHelper.withReadConnectionSource;
 
@@ -239,6 +240,7 @@ private <R> R execute(final BiFunction<Cursor<RawBsonDocument>, OperationContext
             try {
                 return function.apply(wrapped, operationContext);
             } catch (Throwable t) {
+                rethrowIfError(t);
                 if (!isResumableError(t, maxWireVersion)) {
                     throw MongoException.fromThrowableNonNull(t);
                 }

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/ClientSessionPublisherImpl.java

@@ -230,7 +230,9 @@ public Publisher<Void> abortTransaction() {
                 return executor
                         .execute(new AbortTransactionOperation(writeConcern)
                                 .recoveryToken(getRecoveryToken()), readConcern, this)
-                        .onErrorResume(Throwable.class, (e) -> Mono.empty())
+                        .onErrorResume(Exception.class, e ->
+                                // ignore exceptions from abort
+                                Mono.empty())
                         .doOnTerminate(() -> {
                             clearTransactionContext();
                             cleanupTransaction(TransactionState.ABORTED);

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/CommandMarker.java

@@ -87,13 +87,13 @@ class CommandMarker implements Closeable {
     Mono<RawBsonDocument> mark(final String databaseName, final RawBsonDocument command, @Nullable final Timeout operationTimeout) {
         if (client != null) {
             return runCommand(databaseName, command, operationTimeout)
-                    .onErrorResume(Throwable.class, e -> {
+                    .onErrorResume(Exception.class, e -> {
                         if (processBuilder == null || e instanceof MongoOperationTimeoutException) {
                             throw MongoException.fromThrowable(e);
                         }
                         return Mono.fromRunnable(() -> startProcess(processBuilder)).then(runCommand(databaseName, command, operationTimeout));
                     })
-                    .onErrorMap(t -> new MongoClientException("Exception in encryption library: " + t.getMessage(), t));
+                    .onErrorMap(Exception.class, e -> new MongoClientException("Exception in encryption library: " + e.getMessage(), e));
         } else {
             return Mono.fromCallable(() -> command);
         }