diff --git a/CHANGELOG.md b/CHANGELOG.md index ea7ad413a1c..3799be0679d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,3 +27,4 @@ - Trigger an immediate peer search when publishing sync committee messages fails because there are no peers available on the required gossip topic. - Fixed gossip wire validator to reject inbound messages containing the `key` field. - Fixed the gossip message size gate comparing the compressed payload size against the uncompressed `MAX_PAYLOAD_SIZE`. + - Teku now reliably shuts down after an `OutOfMemoryError` instead of continuing to run in a broken state. Out of memory errors wrapped in another exception and async errors are now detected. diff --git a/build.gradle b/build.gradle index 92b9179611e..7902224525e 100644 --- a/build.gradle +++ b/build.gradle @@ -180,12 +180,19 @@ var baseInfrastructureProjects = [ ':infrastructure:bytes', ':infrastructure:collections', ':infrastructure:exceptions', - ':infrastructure:subscribers', ':infrastructure:unsigned', ] dependencyRules { rules { baseInfrastructureProjects.each { register(it) { allowed = [] }} + register(":infrastructure:subscribers") { + // unsigned and logging are the dependencies every non base module gets implicitly + allowed = [ + ":infrastructure:exceptions", + ":infrastructure:logging", + ":infrastructure:unsigned" + ] + } register(":infrastructure:") { allowed = [":infrastructure:"] } diff --git a/infrastructure/async/src/main/java/tech/pegasys/teku/infrastructure/async/SafeFuture.java b/infrastructure/async/src/main/java/tech/pegasys/teku/infrastructure/async/SafeFuture.java index 895725c88fa..78bb006e8cb 100644 --- a/infrastructure/async/src/main/java/tech/pegasys/teku/infrastructure/async/SafeFuture.java +++ b/infrastructure/async/src/main/java/tech/pegasys/teku/infrastructure/async/SafeFuture.java @@ -34,6 +34,7 @@ import org.apache.logging.log4j.Level; import org.apache.logging.log4j.Logger; import tech.pegasys.teku.infrastructure.exceptions.ExceptionUtil; +import tech.pegasys.teku.infrastructure.exceptions.FatalErrorHandler; public class SafeFuture extends CompletableFuture { public static final SafeFuture COMPLETE = SafeFuture.completedFuture(null); @@ -300,6 +301,34 @@ public SafeFuture toVoid() { return thenAccept(__ -> {}); } + /** + * Overridden to detect fatal errors being turned into a failed future. Async code across Teku + * catches {@link Throwable} and converts it into a failed future, which would otherwise allow an + * {@link OutOfMemoryError} to be swallowed and the node to keep running in a broken state. + */ + @Override + public boolean completeExceptionally(final Throwable ex) { + checkForFatalError(ex); + return super.completeExceptionally(ex); + } + + /** + * Shuts the node down if the error is fatal, then returns it unchanged so the normal handling can + * continue. + * + *

Applied to every method that hands an error to user supplied code ({@link + * #exceptionally(Function)}, {@link #handle(BiFunction)}, {@link #whenComplete(BiConsumer)} and + * their variants, which all the other error handling methods are built on). There is no correct + * way to handle a fatal error, so it must trigger a shutdown no matter which handler observes it + * - the vast majority of them just log the error and carry on. + */ + private static Throwable checkForFatalError(final Throwable error) { + if (error != null) { + FatalErrorHandler.shutdownIfFatalError(error, "async task"); + } + return error; + } + public boolean isCompletedNormally() { return isDone() && !isCompletedExceptionally() && !isCancelled(); } @@ -713,20 +742,29 @@ public SafeFuture thenCombineAsync( @Override public SafeFuture exceptionally(final Function fn) { - return (SafeFuture) super.exceptionally(fn); + return (SafeFuture) + super.exceptionally( + error -> FatalErrorHandler.callGuarded(() -> fn.apply(checkForFatalError(error)))); } @SuppressWarnings("unchecked") @Override public SafeFuture handle(final BiFunction fn) { - return (SafeFuture) super.handle(fn); + return (SafeFuture) + super.handle( + (result, error) -> + FatalErrorHandler.callGuarded(() -> fn.apply(result, checkForFatalError(error)))); } @SuppressWarnings("unchecked") @Override public SafeFuture handleAsync( final BiFunction fn, final Executor executor) { - return (SafeFuture) super.handleAsync(fn, executor); + return (SafeFuture) + super.handleAsync( + (result, error) -> + FatalErrorHandler.callGuarded(() -> fn.apply(result, checkForFatalError(error))), + executor); } /** @@ -751,6 +789,7 @@ public SafeFuture handleComposed( try { propagateResult(fn.apply(value, error), result); } catch (final Throwable t) { + checkForFatalError(t); result.completeExceptionally(t); } }); @@ -759,7 +798,11 @@ public SafeFuture handleComposed( @Override public SafeFuture whenComplete(final BiConsumer action) { - return (SafeFuture) super.whenComplete(action); + return (SafeFuture) + super.whenComplete( + (result, error) -> + FatalErrorHandler.runGuarded( + () -> action.accept(result, checkForFatalError(error)))); } public SafeFuture orTimeout(final Duration timeout) { @@ -804,13 +847,12 @@ public SafeFuture orTimeout(final AsyncRunner async, final Duration timeout) * if this future completes exceptionally */ public SafeFuture whenException(final Consumer action) { - return (SafeFuture) - super.whenComplete( - (r, t) -> { - if (t != null) { - action.accept(t); - } - }); + return whenComplete( + (r, t) -> { + if (t != null) { + action.accept(t); + } + }); } /** @@ -818,13 +860,12 @@ public SafeFuture whenException(final Consumer action) { * future completes successfully */ public SafeFuture whenSuccess(final Runnable action) { - return (SafeFuture) - super.whenComplete( - (r, t) -> { - if (t == null) { - action.run(); - } - }); + return whenComplete( + (r, t) -> { + if (t == null) { + action.run(); + } + }); } /** diff --git a/infrastructure/async/src/test/java/tech/pegasys/teku/infrastructure/async/SafeFutureFatalErrorTest.java b/infrastructure/async/src/test/java/tech/pegasys/teku/infrastructure/async/SafeFutureFatalErrorTest.java new file mode 100644 index 00000000000..8037c963363 --- /dev/null +++ b/infrastructure/async/src/test/java/tech/pegasys/teku/infrastructure/async/SafeFutureFatalErrorTest.java @@ -0,0 +1,136 @@ +/* + * Copyright Consensys Software Inc., 2026 + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package tech.pegasys.teku.infrastructure.async; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.logging.log4j.LogManager; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import tech.pegasys.teku.infrastructure.exceptions.ExitConstants; +import tech.pegasys.teku.infrastructure.exceptions.FatalErrorHandler; + +/** + * Async code across Teku catches {@link Throwable} and turns it into a failed future or a log line. + * These tests confirm a fatal error taking those paths still shuts the node down. + */ +@SuppressWarnings("FutureReturnValueIgnored") +class SafeFutureFatalErrorTest { + + private final AtomicInteger terminations = new AtomicInteger(0); + private final List exitCodes = new ArrayList<>(); + + @BeforeEach + void setUp() { + FatalErrorHandler.overrideProcessTerminator( + (exitCode, gracefulTimeout) -> { + terminations.incrementAndGet(); + exitCodes.add(exitCode); + }); + } + + @AfterEach + void tearDown() { + FatalErrorHandler.restoreDefaultProcessTerminator(); + } + + @Test + void shouldShutdownWhenFutureIsCompletedWithOutOfMemoryError() { + new SafeFuture<>().completeExceptionally(new OutOfMemoryError()); + + assertThat(terminations).hasValue(1); + // Restarting is expected to recover from running out of memory + assertThat(exitCodes).containsExactly(ExitConstants.ERROR_EXIT_CODE); + } + + @Test + void shouldShutdownWhenActionThrowsOutOfMemoryError() { + SafeFuture.fromRunnable( + () -> { + throw new OutOfMemoryError(); + }); + + assertThat(terminations).hasValue(1); + } + + @Test + void shouldShutdownWhenOutOfMemoryErrorIsPassedToAnExceptionHandler() { + failedWith(new OutOfMemoryError()).handleException(error -> {}); + + assertThat(terminations).hasValue(1); + } + + @Test + void shouldShutdownWhenOutOfMemoryErrorIsOnlyLogged() { + failedWith(new OutOfMemoryError()).finishError(LogManager.getLogger()); + + assertThat(terminations).hasValue(1); + } + + @Test + void shouldShutdownWhenOutOfMemoryErrorIsReplacedWithAFallbackValue() { + failedWith(new OutOfMemoryError()).exceptionally(error -> null); + + assertThat(terminations).hasValue(1); + } + + @Test + void shouldShutdownWhenOutOfMemoryErrorIsPassedToAFinishErrorHandler() { + failedWith(new OutOfMemoryError()).finish(error -> {}); + + assertThat(terminations).hasValue(1); + } + + @Test + void shouldShutdownWhenOutOfMemoryErrorIsObservedByWhenComplete() { + failedWith(new OutOfMemoryError()).whenComplete((result, error) -> {}); + + assertThat(terminations).hasValue(1); + } + + @Test + void shouldShutdownWhenOutOfMemoryErrorIsNotOneOfTheIgnoredExceptions() { + failedWith(new OutOfMemoryError()).ignoreExceptions(IOException.class); + + assertThat(terminations).hasValue(1); + } + + @Test + void shouldNotShutdownForOrdinaryFailures() { + failedWith(new IOException("nope")).handleException(error -> {}); + failedWith(new IOException("nope")).finishError(LogManager.getLogger()); + failedWith(new IOException("nope")).exceptionally(error -> null); + failedWith(new IOException("nope")).finish(error -> {}); + failedWith(new IOException("nope")).whenComplete((result, error) -> {}); + + assertThat(terminations).hasValue(0); + } + + /** + * Returns a future failed by a task throwing, so the error is wrapped in a {@link + * java.util.concurrent.CompletionException} exactly as it would be in production and doesn't pass + * through {@link SafeFuture#completeExceptionally(Throwable)}. + */ + private SafeFuture failedWith(final Throwable error) { + return SafeFuture.COMPLETE.thenRun( + () -> { + throw new AssertionError("wrapper", error); + }); + } +} diff --git a/infrastructure/exceptions/src/main/java/tech/pegasys/teku/infrastructure/exceptions/FatalErrorHandler.java b/infrastructure/exceptions/src/main/java/tech/pegasys/teku/infrastructure/exceptions/FatalErrorHandler.java new file mode 100644 index 00000000000..f78a0da7c49 --- /dev/null +++ b/infrastructure/exceptions/src/main/java/tech/pegasys/teku/infrastructure/exceptions/FatalErrorHandler.java @@ -0,0 +1,268 @@ +/* + * Copyright Consensys Software Inc., 2026 + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package tech.pegasys.teku.infrastructure.exceptions; + +import com.google.common.annotations.VisibleForTesting; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.Collections; +import java.util.Deque; +import java.util.IdentityHashMap; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Supplier; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +/** + * Detects fatal errors ({@link OutOfMemoryError}) and shuts the process down. + * + *

Once the heap is exhausted the node cannot be trusted to make progress, so the only safe + * option is to exit and let the process supervisor restart us. That only works if we actually + * notice the error, which is harder than it looks: + * + *

    + *
  • The error is almost never seen in its original form. Anything thrown inside an async task + * arrives wrapped in a {@link java.util.concurrent.CompletionException} (possibly several + * layers deep), or as a suppressed error of a combined future's failure, so {@code instanceof + * OutOfMemoryError} checks miss it. {@link #isFatalError} walks the whole cause/suppressed + * graph instead. + *
  • Most of the code base converts any {@link Throwable} into a failed future or a log line and + * carries on. {@link #shutdownIfFatalError(Throwable, String)} is called from those choke + * points so a swallowed error still terminates the node. + *
  • A graceful {@link System#exit(int)} runs shutdown hooks which stop services and wait on + * futures. Under memory pressure that can block forever, leaving a process that is alive but + * useless. We therefore always arm a watchdog which {@link Runtime#halt(int)}s if the + * graceful shutdown doesn't complete in time. + *
+ */ +public class FatalErrorHandler { + private static final Logger LOG = LogManager.getLogger(); + + /** Bounds the cause/suppressed graph traversal, protecting against cycles. */ + private static final int MAX_INSPECTED_ERRORS = 100; + + @VisibleForTesting static final Duration GRACEFUL_SHUTDOWN_TIMEOUT = Duration.ofSeconds(90); + + private static final AtomicBoolean SHUTDOWN_TRIGGERED = new AtomicBoolean(false); + + private static volatile ProcessTerminator processTerminator = FatalErrorHandler::terminateProcess; + + private FatalErrorHandler() {} + + /** + * Runs {@code action}, triggering shutdown if it throws a fatal error, then rethrows. Use this + * for void callbacks (e.g. {@link java.util.function.BiConsumer}) in async pipelines where the + * JVM's {@link java.util.concurrent.CompletableFuture} internals would otherwise swallow the + * error before our {@code completeExceptionally} override can see it. + */ + public static void runGuarded(final Runnable action) { + try { + action.run(); + } catch (final RuntimeException | Error t) { + shutdownIfFatalError(t, "async callback"); + throw t; + } + } + + /** + * Calls {@code action}, triggering shutdown if it throws a fatal error, then rethrows. Use this + * for value-returning callbacks (e.g. {@link java.util.function.Function}) in async pipelines. + */ + public static T callGuarded(final Supplier action) { + try { + return action.get(); + } catch (final RuntimeException | Error t) { + shutdownIfFatalError(t, "async callback"); + throw t; + } + } + + /** + * Returns true if the supplied error is, or was caused by, an unrecoverable error. All causes and + * suppressed errors are inspected, as fatal errors are frequently wrapped by the async framework. + */ + public static boolean isFatalError(final Throwable error) { + // Called for every exceptionally completed future so the common case, a plain chain of causes, + // is checked without allocating anything + Throwable current = error; + for (int depth = 0; current != null && depth < MAX_INSPECTED_ERRORS; depth++) { + if (current instanceof OutOfMemoryError) { + return true; + } + if (current.getSuppressed().length > 0) { + return isFatalErrorInGraph(error); + } + current = current.getCause(); + } + return false; + } + + /** + * Handles the general case where suppressed errors mean causes form a graph rather than a list + */ + private static boolean isFatalErrorInGraph(final Throwable error) { + final Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); + final Deque queue = new ArrayDeque<>(); + queue.add(error); + seen.add(error); + int inspected = 0; + while (!queue.isEmpty() && inspected < MAX_INSPECTED_ERRORS) { + final Throwable current = queue.poll(); + inspected++; + if (current instanceof OutOfMemoryError) { + return true; + } + addIfNotSeen(queue, seen, current.getCause()); + for (Throwable suppressed : current.getSuppressed()) { + addIfNotSeen(queue, seen, suppressed); + } + } + return false; + } + + private static void addIfNotSeen( + final Deque queue, final Set seen, final Throwable error) { + if (error != null && seen.add(error)) { + queue.add(error); + } + } + + /** + * Shuts the node down if the supplied error is or was caused by a fatal error. + * + *

Safe to call from anywhere an error may otherwise be swallowed. Shutdown is only triggered + * once, no matter how many threads report the error. + * + * @param error the error to inspect + * @param context description of where the error was detected, included in the log message + * @return true if the error was fatal, meaning the node is shutting down + */ + public static boolean shutdownIfFatalError(final Throwable error, final String context) { + if (!isFatalError(error)) { + return false; + } + shutdown(error, context); + return true; + } + + /** Shuts the node down, assuming the caller has already established that the error is fatal. */ + public static void shutdown(final Throwable error, final String context) { + if (isShutdownAlreadyTriggered()) { + return; + } + try { + LOG.fatal("Shutting down after unrecoverable error in {}", context, error); + } catch (final Throwable t) { + // Logging itself can fail when out of memory, shutting down still needs to happen + try { + System.err.println( + "Shutting down after unrecoverable error in " + context + " (" + error + ")"); + } catch (final Throwable ignored) { + // String concatenation can also fail under memory pressure; terminate regardless + } + } finally { + processTerminator.terminate(ExitConstants.ERROR_EXIT_CODE, GRACEFUL_SHUTDOWN_TIMEOUT); + } + } + + /** + * Shuts the node down with the specified exit code, for callers that have already reported the + * reason. Unlike {@link System#exit(int)} this never blocks the calling thread and guarantees the + * process actually goes away, even if shutdown hooks get stuck. + */ + public static void terminate(final int exitCode) { + if (isShutdownAlreadyTriggered()) { + return; + } + processTerminator.terminate(exitCode, GRACEFUL_SHUTDOWN_TIMEOUT); + } + + private static boolean isShutdownAlreadyTriggered() { + // Once shutting down, further errors are expected and must not restart the process termination + return !SHUTDOWN_TRIGGERED.compareAndSet(false, true); + } + + private static void terminateProcess(final int exitCode, final Duration gracefulTimeout) { + terminateProcess( + gracefulTimeout, () -> System.exit(exitCode), () -> Runtime.getRuntime().halt(exitCode)); + } + + /** + * Requests a graceful exit, falling back to halting the JVM if it doesn't complete within {@code + * gracefulTimeout}. + * + *

{@link System#exit(int)} blocks the calling thread while shutdown hooks run, so it is + * invoked from a dedicated thread to avoid deadlocking whichever thread reported the error. + * + *

The watchdog is what makes the shutdown reliable. Shutdown hooks stop services, some of + * which wait on latches with no timeout (Jetty's selector shutdown for example), and a hook that + * never returns leaves the JVM alive forever holding the {@code Shutdown} class monitor. {@link + * Runtime#halt(int)} takes a different lock and skips hooks entirely, so it still works in that + * state. + */ + @VisibleForTesting + static void terminateProcess( + final Duration gracefulTimeout, final Runnable exit, final Runnable halt) { + try { + startDaemonThread( + "fatal-error-halt-watchdog", + () -> { + try { + Thread.sleep(gracefulTimeout.toMillis()); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + try { + System.err.println("Graceful shutdown did not complete in time, halting JVM"); + } catch (final Throwable ignored) { + // String writes can fail under memory pressure; halt regardless + } + halt.run(); + }); + startDaemonThread("fatal-error-shutdown", exit); + } catch (final Throwable t) { + // We may not even be able to create threads anymore, so halt without a graceful shutdown + halt.run(); + } + } + + private static void startDaemonThread(final String name, final Runnable action) { + final Thread thread = new Thread(action, name); + thread.setDaemon(true); + thread.start(); + } + + @FunctionalInterface + public interface ProcessTerminator { + void terminate(int exitCode, Duration gracefulTimeout); + } + + /** + * Replaces the process termination logic and clears any previously triggered shutdown. For use by + * tests only, which must restore the default with {@link #restoreDefaultProcessTerminator()}. + */ + @VisibleForTesting + public static void overrideProcessTerminator(final ProcessTerminator terminator) { + processTerminator = terminator; + SHUTDOWN_TRIGGERED.set(false); + } + + @VisibleForTesting + public static void restoreDefaultProcessTerminator() { + processTerminator = FatalErrorHandler::terminateProcess; + SHUTDOWN_TRIGGERED.set(false); + } +} diff --git a/infrastructure/exceptions/src/test/java/tech/pegasys/teku/infrastructure/exceptions/FatalErrorHandlerTest.java b/infrastructure/exceptions/src/test/java/tech/pegasys/teku/infrastructure/exceptions/FatalErrorHandlerTest.java new file mode 100644 index 00000000000..084e031116b --- /dev/null +++ b/infrastructure/exceptions/src/test/java/tech/pegasys/teku/infrastructure/exceptions/FatalErrorHandlerTest.java @@ -0,0 +1,214 @@ +/* + * Copyright Consensys Software Inc., 2026 + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package tech.pegasys.teku.infrastructure.exceptions; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class FatalErrorHandlerTest { + + private final AtomicInteger terminations = new AtomicInteger(0); + private final List exitCodes = new ArrayList<>(); + + @BeforeEach + void setUp() { + FatalErrorHandler.overrideProcessTerminator( + (exitCode, gracefulTimeout) -> { + terminations.incrementAndGet(); + exitCodes.add(exitCode); + }); + } + + @AfterEach + void tearDown() { + FatalErrorHandler.restoreDefaultProcessTerminator(); + } + + @Test + void isFatalError_shouldDetectDirectOutOfMemoryError() { + assertThat(FatalErrorHandler.isFatalError(new OutOfMemoryError("Java heap space"))).isTrue(); + } + + @Test + void isFatalError_shouldDetectOutOfMemoryErrorWrappedByAsyncFramework() { + final Throwable error = + new CompletionException( + new ExecutionException(new IllegalStateException(new OutOfMemoryError()))); + + assertThat(FatalErrorHandler.isFatalError(error)).isTrue(); + } + + @Test + void isFatalError_shouldDetectSuppressedOutOfMemoryError() { + // SafeFuture.allOf reports the additional failures as suppressed errors + final Throwable error = new CompletionException(new IOException("first failure")); + error.addSuppressed(new OutOfMemoryError()); + + assertThat(FatalErrorHandler.isFatalError(error)).isTrue(); + } + + @Test + void isFatalError_shouldDetectSubclassesOfOutOfMemoryError() { + // For example Netty's OutOfDirectMemoryError + assertThat(FatalErrorHandler.isFatalError(new RuntimeException(new CustomOutOfMemoryError()))) + .isTrue(); + } + + @Test + void isFatalError_shouldNotDetectOrdinaryErrors() { + assertThat(FatalErrorHandler.isFatalError(new CompletionException(new IOException("nope")))) + .isFalse(); + assertThat(FatalErrorHandler.isFatalError(new IllegalStateException())).isFalse(); + assertThat(FatalErrorHandler.isFatalError(null)).isFalse(); + } + + @Test + void isFatalError_shouldNotLoopForeverWhenCausesAreCyclic() { + final RuntimeException a = new RuntimeException(); + final RuntimeException b = new RuntimeException(a); + a.addSuppressed(b); + + assertThat(FatalErrorHandler.isFatalError(b)).isFalse(); + } + + @Test + void shutdownIfFatalError_shouldTerminateWithErrorExitCodeWhenFatal() { + assertThat( + FatalErrorHandler.shutdownIfFatalError( + new CompletionException(new OutOfMemoryError()), "test")) + .isTrue(); + + assertThat(terminations).hasValue(1); + assertThat(exitCodes).containsExactly(ExitConstants.ERROR_EXIT_CODE); + } + + @Test + void shutdownIfFatalError_shouldDoNothingWhenNotFatal() { + assertThat(FatalErrorHandler.shutdownIfFatalError(new IOException("nope"), "test")).isFalse(); + + assertThat(terminations).hasValue(0); + } + + @Test + void shutdownIfFatalError_shouldOnlyTerminateOnceWhenMultipleErrorsAreReported() { + for (int i = 0; i < 5; i++) { + assertThat(FatalErrorHandler.shutdownIfFatalError(new OutOfMemoryError(), "test")).isTrue(); + } + + assertThat(terminations).hasValue(1); + } + + @Test + void terminate_shouldUseTheSuppliedExitCode() { + FatalErrorHandler.terminate(ExitConstants.FATAL_EXIT_CODE); + + assertThat(exitCodes).containsExactly(ExitConstants.FATAL_EXIT_CODE); + } + + @Test + void terminate_shouldOnlyTerminateOnce() { + FatalErrorHandler.terminate(ExitConstants.FATAL_EXIT_CODE); + FatalErrorHandler.terminate(ExitConstants.ERROR_EXIT_CODE); + FatalErrorHandler.shutdownIfFatalError(new OutOfMemoryError(), "test"); + + assertThat(terminations).hasValue(1); + } + + /** + * Reproduces the shape of a real hang: an out of memory error triggered {@link System#exit(int)}, + * which ran the shutdown hook, which blocked forever stopping the REST API (Jetty waits on an + * untimed latch). The JVM stayed alive for days with every other thread that called exit blocked + * on the {@code Shutdown} class monitor. + */ + @Test + void terminateProcess_shouldHaltWhenAShutdownHookNeverReturns() throws Exception { + final CountDownLatch halted = new CountDownLatch(1); + final CountDownLatch wedgedHook = new CountDownLatch(1); + + FatalErrorHandler.terminateProcess( + Duration.ofMillis(100), + () -> { + // System.exit never returns because a shutdown hook is stuck + try { + wedgedHook.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + }, + halted::countDown); + + assertThat(halted.await(30, TimeUnit.SECONDS)).isTrue(); + wedgedHook.countDown(); + } + + @Test + void terminateProcess_shouldNotBlockTheReportingThread() { + final CountDownLatch wedgedHook = new CountDownLatch(1); + + // Returns rather than blocking with the wedged exit call, unlike calling System.exit directly + FatalErrorHandler.terminateProcess( + Duration.ofHours(1), + () -> { + try { + wedgedHook.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + }, + () -> {}); + + wedgedHook.countDown(); + } + + @Test + void terminateProcess_shouldUseDaemonThreadsSoTheyCannotKeepTheJvmAlive() { + FatalErrorHandler.terminateProcess(Duration.ofHours(1), () -> {}, () -> {}); + + assertThat(findThreads("fatal-error-")) + .isNotEmpty() + .allSatisfy(thread -> assertThat(thread.isDaemon()).isTrue()); + } + + private List findThreads(final String namePrefix) { + return Thread.getAllStackTraces().keySet().stream() + .filter(thread -> thread.getName().startsWith(namePrefix)) + .toList(); + } + + @Test + void shutdown_shouldPassAGracefulShutdownTimeoutToTheTerminator() { + final List timeouts = new ArrayList<>(); + FatalErrorHandler.overrideProcessTerminator( + (exitCode, gracefulTimeout) -> timeouts.add(gracefulTimeout)); + + FatalErrorHandler.shutdown(new OutOfMemoryError(), "test"); + + assertThat(timeouts).containsExactly(FatalErrorHandler.GRACEFUL_SHUTDOWN_TIMEOUT); + } + + // Static so that this Serializable class doesn't reference the non serializable test class + private static class CustomOutOfMemoryError extends OutOfMemoryError {} +} diff --git a/infrastructure/restapi/src/main/java/tech/pegasys/teku/infrastructure/restapi/DefaultExceptionHandler.java b/infrastructure/restapi/src/main/java/tech/pegasys/teku/infrastructure/restapi/DefaultExceptionHandler.java index 59f433407b3..b0b6f285857 100644 --- a/infrastructure/restapi/src/main/java/tech/pegasys/teku/infrastructure/restapi/DefaultExceptionHandler.java +++ b/infrastructure/restapi/src/main/java/tech/pegasys/teku/infrastructure/restapi/DefaultExceptionHandler.java @@ -22,6 +22,7 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import tech.pegasys.teku.infrastructure.exceptions.ExceptionUtil; +import tech.pegasys.teku.infrastructure.exceptions.FatalErrorHandler; import tech.pegasys.teku.infrastructure.http.HttpErrorResponse; import tech.pegasys.teku.infrastructure.json.JsonUtil; @@ -30,6 +31,9 @@ public class DefaultExceptionHandler implements ExceptionHa @Override public void handle(final T throwable, final Context context) { + // Requests must not be able to swallow a fatal error and leave the node running + FatalErrorHandler.shutdownIfFatalError(throwable, "REST API request"); + if (ExceptionUtil.hasCause(throwable, EOFException.class)) { LOG.trace("Connection closed before response could be completed.", throwable); return; diff --git a/infrastructure/subscribers/build.gradle b/infrastructure/subscribers/build.gradle index c773c5128a3..0b155353d17 100644 --- a/infrastructure/subscribers/build.gradle +++ b/infrastructure/subscribers/build.gradle @@ -1 +1,3 @@ -dependencies {} +dependencies { + implementation project(":infrastructure:exceptions") +} diff --git a/infrastructure/subscribers/src/main/java/tech/pegasys/teku/infrastructure/subscribers/ObservableValue.java b/infrastructure/subscribers/src/main/java/tech/pegasys/teku/infrastructure/subscribers/ObservableValue.java index c88cc7d83f5..494e85ce00a 100644 --- a/infrastructure/subscribers/src/main/java/tech/pegasys/teku/infrastructure/subscribers/ObservableValue.java +++ b/infrastructure/subscribers/src/main/java/tech/pegasys/teku/infrastructure/subscribers/ObservableValue.java @@ -19,6 +19,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import tech.pegasys.teku.infrastructure.exceptions.FatalErrorHandler; /** * A value holder class which notifies subscribers on value updates @@ -121,7 +122,10 @@ private void notify(final Subscription subscription, final C value) { subscription.getSubscriber().onValueChanged(value); } catch (Throwable throwable) { if (suppressCallbackExceptions) { - LOG.error("Error in callback: ", throwable); + // A fatal error must shut the node down rather than just being logged + if (!FatalErrorHandler.shutdownIfFatalError(throwable, "value change callback")) { + LOG.error("Error in callback: ", throwable); + } } else { throw throwable; } diff --git a/infrastructure/subscribers/src/main/java/tech/pegasys/teku/infrastructure/subscribers/Subscribers.java b/infrastructure/subscribers/src/main/java/tech/pegasys/teku/infrastructure/subscribers/Subscribers.java index 5b583396811..96225d7f7a3 100644 --- a/infrastructure/subscribers/src/main/java/tech/pegasys/teku/infrastructure/subscribers/Subscribers.java +++ b/infrastructure/subscribers/src/main/java/tech/pegasys/teku/infrastructure/subscribers/Subscribers.java @@ -20,6 +20,7 @@ import java.util.function.Consumer; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import tech.pegasys.teku.infrastructure.exceptions.FatalErrorHandler; /** * Tracks subscribers that should be notified when some event occurred. This class is safe to use @@ -98,7 +99,10 @@ public void forEach(final Consumer action) { action.accept(subscriber); } catch (Throwable throwable) { if (suppressCallbackExceptions) { - LOG.error("Error in callback: ", throwable); + // A fatal error must shut the node down rather than just being logged + if (!FatalErrorHandler.shutdownIfFatalError(throwable, "subscriber callback")) { + LOG.error("Error in callback: ", throwable); + } } else { throw throwable; } diff --git a/infrastructure/subscribers/src/test/java/tech/pegasys/teku/infrastructure/subscribers/ObservableValueTest.java b/infrastructure/subscribers/src/test/java/tech/pegasys/teku/infrastructure/subscribers/ObservableValueTest.java index a8ff748c0af..09f62e80b77 100644 --- a/infrastructure/subscribers/src/test/java/tech/pegasys/teku/infrastructure/subscribers/ObservableValueTest.java +++ b/infrastructure/subscribers/src/test/java/tech/pegasys/teku/infrastructure/subscribers/ObservableValueTest.java @@ -20,10 +20,51 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Stream; import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import tech.pegasys.teku.infrastructure.exceptions.ExitConstants; +import tech.pegasys.teku.infrastructure.exceptions.FatalErrorHandler; public class ObservableValueTest { + @AfterEach + void restoreFatalErrorHandler() { + FatalErrorHandler.restoreDefaultProcessTerminator(); + } + + // Suppressing exceptions must not hide the node running out of memory + @Test + public void shouldShutdownWhenASuppressedCallbackFailsFatally() { + final List exitCodes = new ArrayList<>(); + FatalErrorHandler.overrideProcessTerminator( + (exitCode, gracefulTimeout) -> exitCodes.add(exitCode)); + final ObservableValue observableValue = new ObservableValue<>(true); + observableValue.subscribe( + value -> { + throw new IllegalStateException("whoops", new OutOfMemoryError()); + }); + + observableValue.set("value"); + + Assertions.assertThat(exitCodes).containsExactly(ExitConstants.ERROR_EXIT_CODE); + } + + @Test + public void shouldNotShutdownWhenASuppressedCallbackFailsWithAnOrdinaryException() { + final List exitCodes = new ArrayList<>(); + FatalErrorHandler.overrideProcessTerminator( + (exitCode, gracefulTimeout) -> exitCodes.add(exitCode)); + final ObservableValue observableValue = new ObservableValue<>(true); + observableValue.subscribe( + value -> { + throw new IllegalStateException("whoops"); + }); + + observableValue.set("value"); + + Assertions.assertThat(exitCodes).isEmpty(); + } + @Test public void testConcurrentSubscribersNotifications() throws InterruptedException { ObservableValue observableValue = new ObservableValue<>(false); diff --git a/infrastructure/subscribers/src/test/java/tech/pegasys/teku/infrastructure/subscribers/SubscribersTest.java b/infrastructure/subscribers/src/test/java/tech/pegasys/teku/infrastructure/subscribers/SubscribersTest.java index cc4ecfea7f6..4397fddc9f8 100644 --- a/infrastructure/subscribers/src/test/java/tech/pegasys/teku/infrastructure/subscribers/SubscribersTest.java +++ b/infrastructure/subscribers/src/test/java/tech/pegasys/teku/infrastructure/subscribers/SubscribersTest.java @@ -20,13 +20,32 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; +import java.util.ArrayList; +import java.util.List; import java.util.function.Consumer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import tech.pegasys.teku.infrastructure.exceptions.ExitConstants; +import tech.pegasys.teku.infrastructure.exceptions.FatalErrorHandler; public class SubscribersTest { + private final Runnable subscriber1 = mock(Runnable.class); private final Runnable subscriber2 = mock(Runnable.class); private final Subscribers subscribers = Subscribers.create(false); + private final List exitCodes = new ArrayList<>(); + + @BeforeEach + void setUpFatalErrorHandler() { + FatalErrorHandler.overrideProcessTerminator( + (exitCode, gracefulTimeout) -> exitCodes.add(exitCode)); + } + + @AfterEach + void restoreFatalErrorHandler() { + FatalErrorHandler.restoreDefaultProcessTerminator(); + } @Test public void shouldAddSubscriber() { @@ -85,6 +104,21 @@ public void suppressCallbackExceptions_true() { // No Exception should be thrown subscribers.forEach(Runnable::run); + + assertThat(exitCodes).isEmpty(); + } + + // Suppressing exceptions must not hide the node running out of memory + @Test + public void suppressCallbackExceptions_shouldStillShutdownOnFatalError() { + final Subscribers subscribers = Subscribers.create(true); + + doThrow(new IllegalStateException("whoops", new OutOfMemoryError())).when(subscriber1).run(); + subscribers.subscribe(subscriber1); + + subscribers.forEach(Runnable::run); + + assertThat(exitCodes).containsExactly(ExitConstants.ERROR_EXIT_CODE); } @SuppressWarnings("unchecked") diff --git a/teku/src/main/java/tech/pegasys/teku/TekuDefaultExceptionHandler.java b/teku/src/main/java/tech/pegasys/teku/TekuDefaultExceptionHandler.java index d93bc035bfd..232f80eecd7 100644 --- a/teku/src/main/java/tech/pegasys/teku/TekuDefaultExceptionHandler.java +++ b/teku/src/main/java/tech/pegasys/teku/TekuDefaultExceptionHandler.java @@ -27,6 +27,7 @@ import org.apache.logging.log4j.Logger; import tech.pegasys.teku.infrastructure.events.ChannelExceptionHandler; import tech.pegasys.teku.infrastructure.exceptions.ExceptionUtil; +import tech.pegasys.teku.infrastructure.exceptions.FatalErrorHandler; import tech.pegasys.teku.infrastructure.exceptions.FatalServiceFailureException; import tech.pegasys.teku.infrastructure.logging.StatusLogger; import tech.pegasys.teku.services.beaconchain.EphemeryLifecycleException; @@ -76,19 +77,21 @@ private void handleException(final Throwable exception, final String subscriberD if (fatalServiceError.isPresent()) { final String failedService = fatalServiceError.get().getService(); - statusLog.fatalError(failedService, exception); - System.exit(FATAL_EXIT_CODE); + tryLog(() -> statusLog.fatalError(failedService, exception)); + FatalErrorHandler.terminate(FATAL_EXIT_CODE); } else if (ExceptionUtil.getCause(exception, DatabaseStorageException.class) .filter(DatabaseStorageException::isUnrecoverable) .isPresent()) { - statusLog.fatalError(subscriberDescription, exception); - System.exit(FATAL_EXIT_CODE); - } else if (exception instanceof OutOfMemoryError) { - statusLog.fatalError(subscriberDescription, exception); - System.exit(ERROR_EXIT_CODE); + tryLog(() -> statusLog.fatalError(subscriberDescription, exception)); + FatalErrorHandler.terminate(FATAL_EXIT_CODE); + } else if (FatalErrorHandler.isFatalError(exception)) { + // An out of memory error is frequently wrapped, so all causes are checked. Exits with + // ERROR_EXIT_CODE because restarting is expected to recover. + tryLog(() -> statusLog.fatalError(subscriberDescription, exception)); + FatalErrorHandler.terminate(ERROR_EXIT_CODE); } else if (exception instanceof EphemeryLifecycleException) { - statusLog.fatalError(subscriberDescription, exception); - System.exit(ERROR_EXIT_CODE); + tryLog(() -> statusLog.fatalError(subscriberDescription, exception)); + FatalErrorHandler.terminate(ERROR_EXIT_CODE); } else if (exception instanceof ShuttingDownException) { LOG.debug("Shutting down", exception); } else if (isExpectedNettyError(exception)) { @@ -103,6 +106,15 @@ private void handleException(final Throwable exception, final String subscriberD } } + @SuppressWarnings("unused") + private static void tryLog(final Runnable logAction) { + try { + logAction.run(); + } catch (final Throwable ignored) { + // Logging may itself throw when resources (e.g. memory) are exhausted; proceed to terminate. + } + } + private boolean isExpectedNettyError(final Throwable exception) { return exception instanceof ClosedChannelException; } diff --git a/teku/src/test/java/tech/pegasys/teku/TekuDefaultExceptionHandlerTest.java b/teku/src/test/java/tech/pegasys/teku/TekuDefaultExceptionHandlerTest.java index 1083ddc0701..33e07deac7c 100644 --- a/teku/src/test/java/tech/pegasys/teku/TekuDefaultExceptionHandlerTest.java +++ b/teku/src/test/java/tech/pegasys/teku/TekuDefaultExceptionHandlerTest.java @@ -13,19 +13,44 @@ package tech.pegasys.teku; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletionException; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import tech.pegasys.teku.infrastructure.exceptions.ExitConstants; +import tech.pegasys.teku.infrastructure.exceptions.FatalErrorHandler; +import tech.pegasys.teku.infrastructure.exceptions.FatalServiceFailureException; import tech.pegasys.teku.infrastructure.logging.StatusLogger; +import tech.pegasys.teku.services.beaconchain.EphemeryLifecycleException; +import tech.pegasys.teku.storage.server.DatabaseStorageException; class TekuDefaultExceptionHandlerTest { private final StatusLogger log = mock(StatusLogger.class); private final TekuDefaultExceptionHandler exceptionHandler = new TekuDefaultExceptionHandler(log); + private final List exitCodes = new ArrayList<>(); + + @BeforeEach + void setUp() { + FatalErrorHandler.overrideProcessTerminator( + (exitCode, gracefulTimeout) -> exitCodes.add(exitCode)); + } + + @AfterEach + void tearDown() { + FatalErrorHandler.restoreDefaultProcessTerminator(); + } @Test void logWarningIfAssertFails() { @@ -43,4 +68,80 @@ void logFatalIfNonAssertExceptionThrown() { verify(log).unexpectedFailure(anyString(), eq(exception)); } + + @Test + void shouldNotShutdownForOrdinaryFailures() { + exceptionHandler.uncaughtException(Thread.currentThread(), new IOException("nope")); + + assertThat(exitCodes).isEmpty(); + } + + // Out of memory is expected to be recoverable by a restart, so it must not use the exit code + // which tells operators not to restart + @Test + void shouldExitWithErrorExitCodeWhenOutOfMemoryErrorIsThrown() { + final OutOfMemoryError error = new OutOfMemoryError(); + + exceptionHandler.uncaughtException(Thread.currentThread(), error); + + verify(log).fatalError(anyString(), eq(error)); + assertThat(exitCodes).containsExactly(ExitConstants.ERROR_EXIT_CODE); + } + + @Test + void shouldExitWithErrorExitCodeWhenOutOfMemoryErrorIsWrappedInAnotherException() { + final Throwable error = + new CompletionException(new IllegalArgumentException(new OutOfMemoryError())); + + exceptionHandler.uncaughtException(Thread.currentThread(), error); + + verify(log).fatalError(anyString(), eq(error)); + verify(log, never()).specificationFailure(anyString(), any()); + assertThat(exitCodes).containsExactly(ExitConstants.ERROR_EXIT_CODE); + } + + @Test + void shouldExitWithFatalExitCodeWhenAServiceFailsFatally() { + final Throwable error = + new CompletionException( + new FatalServiceFailureException( + TekuDefaultExceptionHandlerTest.class, new IOException())); + + exceptionHandler.uncaughtException(Thread.currentThread(), error); + + assertThat(exitCodes).containsExactly(ExitConstants.FATAL_EXIT_CODE); + } + + @Test + void shouldExitWithFatalExitCodeWhenStorageIsUnrecoverable() { + final Throwable error = + new CompletionException( + DatabaseStorageException.unrecoverable("corrupt", new IOException())); + + exceptionHandler.uncaughtException(Thread.currentThread(), error); + + assertThat(exitCodes).containsExactly(ExitConstants.FATAL_EXIT_CODE); + } + + // A fatal service failure means a restart won't help, even when it was caused by running out of + // memory, so it keeps precedence over the out of memory check + @Test + void shouldExitWithFatalExitCodeWhenAFatalServiceFailureWasCausedByOutOfMemory() { + final Throwable error = + new CompletionException( + new FatalServiceFailureException( + TekuDefaultExceptionHandlerTest.class, new OutOfMemoryError())); + + exceptionHandler.uncaughtException(Thread.currentThread(), error); + + assertThat(exitCodes).containsExactly(ExitConstants.FATAL_EXIT_CODE); + } + + @Test + void shouldExitWithErrorExitCodeWhenEphemeryLifecycleEnds() { + exceptionHandler.uncaughtException( + Thread.currentThread(), new EphemeryLifecycleException("reset required")); + + assertThat(exitCodes).containsExactly(ExitConstants.ERROR_EXIT_CODE); + } }