Ensure no circular reference in translog tragic exception (#55959)

We generate a circular reference exception in translog in 6.8 in the
following scenario:

- The first rollGeneration hits "too many open files" exception when it's
copying a checkpoint file. We will set the tragic exception and close
the translog

- The second rollGeneration hits AlreadyClosedException as the current
writer is closed. We will suppress the ACE to the current tragic
exception. Unfortunately, this leads to a circular reference as ACE
already suppresses the tragic exception.

Other factors that help to manifest this bug:
- We do not fail the engine on AlreadyClosedException in flush
- We do not check for ensureOpen before rolling a new generation

Closes #55893

Author: dnhatn

server/src/main/java/org/elasticsearch/ExceptionsHelper.java

@@ -38,12 +38,14 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.IdentityHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Queue;
 import java.util.Set;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 public final class ExceptionsHelper {
@@ -252,6 +254,31 @@ public static Optional<Error> maybeError(final Throwable cause, final Logger log
         return Optional.empty();
     }
 
+    @SuppressWarnings("unchecked")
+    public static <T extends Throwable> Optional<T> unwrapCausesAndSuppressed(Throwable cause, Predicate<Throwable> predicate) {
+        if (predicate.test(cause)) {
+            return Optional.of((T) cause);
+        }
+
+        final Queue<Throwable> queue = new LinkedList<>();
+        queue.add(cause);
+        final Set<Throwable> seen = Collections.newSetFromMap(new IdentityHashMap<>());
+        while (queue.isEmpty() == false) {
+            final Throwable current = queue.remove();
+            if (seen.add(current) == false) {
+                continue;
+            }
+            if (predicate.test(current)) {
+                return Optional.of((T) current);
+            }
+            Collections.addAll(queue, current.getSuppressed());
+            if (current.getCause() != null) {
+                queue.add(current.getCause());
+            }
+        }
+        return Optional.empty();
+    }
+
     /**
      * See {@link #maybeError(Throwable, Logger)}. Uses the class-local logger.
      */

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -1838,6 +1838,7 @@ public CommitId flush(boolean force, boolean waitIfOngoing) throws EngineExcepti
                         refresh("version_table_flush", SearcherScope.INTERNAL);
                         translog.trimUnreferencedReaders();
                     } catch (AlreadyClosedException e) {
+                        failOnTragicEvent(e);
                         throw e;
                     } catch (Exception e) {
                         throw new FlushFailedEngineException(shardId, e);

server/src/main/java/org/elasticsearch/index/translog/TragicExceptionHolder.java

@@ -19,6 +19,9 @@
 
 package org.elasticsearch.index.translog;
 
+import org.apache.lucene.store.AlreadyClosedException;
+import org.elasticsearch.ExceptionsHelper;
+
 import java.util.concurrent.atomic.AtomicReference;
 
 public class TragicExceptionHolder {
@@ -30,10 +33,15 @@ public class TragicExceptionHolder {
      */
     public void setTragicException(Exception ex) {
         assert ex != null;
-        if (tragedy.compareAndSet(null, ex) == false) {
-            if (tragedy.get() != ex) { // to ensure there is no self-suppression
-                tragedy.get().addSuppressed(ex);
-            }
+        if (tragedy.compareAndSet(null, ex)) {
+            return; // first exception
+        }
+        final Exception tragedy = this.tragedy.get();
+        // ensure no circular reference
+        if (ExceptionsHelper.unwrapCausesAndSuppressed(ex, e -> e == tragedy).isPresent()) {
+            assert ex == tragedy || ex instanceof AlreadyClosedException : new AssertionError("must be ACE or tragic exception", ex);
+        } else {
+            tragedy.addSuppressed(ex);
         }
     }
 

server/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -1702,6 +1702,7 @@ public TranslogGeneration getMinGenerationForSeqNo(final long seqNo) {
      */
     public void rollGeneration() throws IOException {
         try (Releasable ignored = writeLock.acquire()) {
+            ensureOpen();
             try {
                 final TranslogReader reader = current.closeIntoReader();
                 readers.add(reader);

server/src/test/java/org/elasticsearch/index/translog/TranslogTests.java

@@ -98,9 +98,12 @@
 import java.util.Deque;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.IdentityHashMap;
 import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
@@ -3270,4 +3273,58 @@ public void copy(Path source, Path target, CopyOption... options) throws IOExcep
             }
         }
     }
+
+    public void testEnsureNoCircularException() throws Exception {
+        final AtomicBoolean failedToSyncCheckpoint = new AtomicBoolean();
+        final ChannelFactory channelFactory = (file, openOption) -> {
+            final FileChannel channel = FileChannel.open(file, openOption);
+            return new FilterFileChannel(channel) {
+                @Override
+                public void force(boolean metaData) throws IOException {
+                    if (failedToSyncCheckpoint.get()) {
+                        throw new IOException("simulated");
+                    }
+                    super.force(metaData);
+                }
+            };
+        };
+        final TranslogConfig config = getTranslogConfig(createTempDir());
+        final String translogUUID = Translog.createEmptyTranslog(
+            config.getTranslogPath(), SequenceNumbers.NO_OPS_PERFORMED, shardId, channelFactory, primaryTerm.get());
+        final Translog translog = new Translog(config, translogUUID, createTranslogDeletionPolicy(),
+            () -> SequenceNumbers.NO_OPS_PERFORMED, primaryTerm::get) {
+            @Override
+            ChannelFactory getChannelFactory() {
+                return channelFactory;
+            }
+        };
+        try {
+            translog.add(new Translog.Index("1", "_doc", 1, primaryTerm.get(), new byte[]{1}));
+            failedToSyncCheckpoint.set(true);
+            expectThrows(IOException.class, translog::rollGeneration);
+            final AlreadyClosedException alreadyClosedException = expectThrows(AlreadyClosedException.class, translog::rollGeneration);
+            if (hasCircularReference(alreadyClosedException)) {
+                throw new AssertionError("detect circular reference exception", alreadyClosedException);
+            }
+        } finally {
+            IOUtils.close(translog);
+        }
+    }
+
+    static boolean hasCircularReference(Exception cause) {
+        final Queue<Throwable> queue = new LinkedList<>();
+        queue.add(cause);
+        final Set<Throwable> seen = Collections.newSetFromMap(new IdentityHashMap<>());
+        while (queue.isEmpty() == false) {
+            final Throwable current = queue.remove();
+            if (seen.add(current) == false) {
+                return true;
+            }
+            Collections.addAll(queue, current.getSuppressed());
+            if (current.getCause() != null) {
+                queue.add(current.getCause());
+            }
+        }
+        return false;
+    }
 }