Fix: OOM didn't force Teku to quit

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.

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:"]
 		}

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<T> extends CompletableFuture<T> {
   public static final SafeFuture<Void> COMPLETE = SafeFuture.completedFuture(null);
@@ -300,6 +301,34 @@ public SafeFuture<Void> 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.
+   *
+   * <p>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 <U, V> SafeFuture<V> thenCombineAsync(
 
   @Override
   public SafeFuture<T> exceptionally(final Function<Throwable, ? extends T> fn) {
-    return (SafeFuture<T>) super.exceptionally(fn);
+    return (SafeFuture<T>)
+        super.exceptionally(
+            error -> FatalErrorHandler.callGuarded(() -> fn.apply(checkForFatalError(error))));
   }
 
   @SuppressWarnings("unchecked")
   @Override
   public <U> SafeFuture<U> handle(final BiFunction<? super T, Throwable, ? extends U> fn) {
-    return (SafeFuture<U>) super.handle(fn);
+    return (SafeFuture<U>)
+        super.handle(
+            (result, error) ->
+                FatalErrorHandler.callGuarded(() -> fn.apply(result, checkForFatalError(error))));
   }
 
   @SuppressWarnings("unchecked")
   @Override
   public <U> SafeFuture<U> handleAsync(
       final BiFunction<? super T, Throwable, ? extends U> fn, final Executor executor) {
-    return (SafeFuture<U>) super.handleAsync(fn, executor);
+    return (SafeFuture<U>)
+        super.handleAsync(
+            (result, error) ->
+                FatalErrorHandler.callGuarded(() -> fn.apply(result, checkForFatalError(error))),
+            executor);
   }
 
   /**
@@ -751,6 +789,7 @@ public <U> SafeFuture<U> handleComposed(
           try {
             propagateResult(fn.apply(value, error), result);
           } catch (final Throwable t) {
+            checkForFatalError(t);
             result.completeExceptionally(t);
           }
         });
@@ -759,7 +798,11 @@ public <U> SafeFuture<U> handleComposed(
 
   @Override
   public SafeFuture<T> whenComplete(final BiConsumer<? super T, ? super Throwable> action) {
-    return (SafeFuture<T>) super.whenComplete(action);
+    return (SafeFuture<T>)
+        super.whenComplete(
+            (result, error) ->
+                FatalErrorHandler.runGuarded(
+                    () -> action.accept(result, checkForFatalError(error))));
   }
 
   public SafeFuture<T> orTimeout(final Duration timeout) {
@@ -804,27 +847,25 @@ public SafeFuture<T> orTimeout(final AsyncRunner async, final Duration timeout)
    * if this future completes exceptionally
    */
   public SafeFuture<T> whenException(final Consumer<Throwable> action) {
-    return (SafeFuture<T>)
-        super.whenComplete(
-            (r, t) -> {
-              if (t != null) {
-                action.accept(t);
-              }
-            });
+    return whenComplete(
+        (r, t) -> {
+          if (t != null) {
+            action.accept(t);
+          }
+        });
   }
 
   /**
    * Returns the future which completes with the same result or exception. The action is run if this
    * future completes successfully
    */
   public SafeFuture<T> whenSuccess(final Runnable action) {
-    return (SafeFuture<T>)
-        super.whenComplete(
-            (r, t) -> {
-              if (t == null) {
-                action.run();
-              }
-            });
+    return whenComplete(
+        (r, t) -> {
+          if (t == null) {
+            action.run();
+          }
+        });
   }
 
   /**

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<Integer> 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<Void> failedWith(final Throwable error) {
+    return SafeFuture.COMPLETE.thenRun(
+        () -> {
+          throw new AssertionError("wrapper", error);
+        });
+  }
+}