Store Pending Deletions Fix (#40345)

FilterDirectory.getPendingDeletions does not delegate, fixed
temporarily by overriding in StoreDirectory.

This in turn caused duplicate file name use after a trimUnsafeCommits
had been done, since a new IndexWriter would not consider the pending
deletes in IndexFileDeleter. This should only happen on windows (AFAIK).

Reenabled doing index updates for all tests using
IndexShardTests.indexOnReplicaWithGaps (which could fail due to above
when using mocked WindowsFS).

Added getPendingDeletions delegation to all elasticsearch
FilterDirectory subclasses that were not trivial test-only overrides to
minimize the risk of hitting this issue in another case.

Author: henningandersen

plugins/store-smb/src/main/java/org/elasticsearch/index/store/SmbDirectoryWrapper.java

@@ -19,17 +19,19 @@
 
 package org.elasticsearch.index.store;
 
-import java.io.FilterOutputStream;
-import java.io.IOException;
-import java.nio.channels.Channels;
-import java.nio.file.Files;
-import java.nio.file.StandardOpenOption;
 import org.apache.lucene.store.FSDirectory;
 import org.apache.lucene.store.FilterDirectory;
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.store.OutputStreamIndexOutput;
 
+import java.io.FilterOutputStream;
+import java.io.IOException;
+import java.nio.channels.Channels;
+import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
+import java.util.Set;
+
 /**
  * This class is used to wrap an existing {@link org.apache.lucene.store.FSDirectory} so that
  * the new shard segment files will be opened for Read and Write access.
@@ -78,4 +80,10 @@ public void write(byte[] b, int offset, int length) throws IOException {
                     CHUNK_SIZE);
         }
     }
+
+    // temporary override until LUCENE-8735 is integrated
+    @Override
+    public Set<String> getPendingDeletions() throws IOException {
+        return in.getPendingDeletions();
+    }
 }

server/src/main/java/org/elasticsearch/index/shard/LocalShardSnapshot.java

@@ -33,6 +33,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Collection;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 final class LocalShardSnapshot implements Closeable {
@@ -116,6 +117,12 @@ public Lock obtainLock(String name) throws IOException {
             public void close() throws IOException {
                 throw new UnsupportedOperationException("nobody should close this directory wrapper");
             }
+
+            // temporary override until LUCENE-8735 is integrated
+            @Override
+            public Set<String> getPendingDeletions() throws IOException {
+                return in.getPendingDeletions();
+            }
         };
     }
 

server/src/main/java/org/elasticsearch/index/shard/StoreRecovery.java

@@ -261,6 +261,12 @@ public void readBytes(byte[] b, int offset, int len) throws IOException {
                 assert index.getFileDetails(dest).recovered() == l : index.getFileDetails(dest).toString();
             }
         }
+
+        // temporary override until LUCENE-8735 is integrated
+        @Override
+        public Set<String> getPendingDeletions() throws IOException {
+            return in.getPendingDeletions();
+        }
     }
 
     /**

server/src/main/java/org/elasticsearch/index/store/ByteSizeCachingDirectory.java

@@ -32,6 +32,7 @@
 import java.io.UncheckedIOException;
 import java.nio.file.AccessDeniedException;
 import java.nio.file.NoSuchFileException;
+import java.util.Set;
 
 final class ByteSizeCachingDirectory extends FilterDirectory {
 
@@ -180,4 +181,9 @@ public void deleteFile(String name) throws IOException {
         }
     }
 
+    // temporary override until LUCENE-8735 is integrated
+    @Override
+    public Set<String> getPendingDeletions() throws IOException {
+        return in.getPendingDeletions();
+    }
 }

server/src/main/java/org/elasticsearch/index/store/Store.java

@@ -734,6 +734,13 @@ private void innerClose() throws IOException {
         public String toString() {
             return "store(" + in.toString() + ")";
         }
+
+        @Override
+        public Set<String> getPendingDeletions() throws IOException {
+            // FilterDirectory.getPendingDeletions does not delegate, working around it here.
+            // to be removed once fixed in FilterDirectory.
+            return unwrap(this).getPendingDeletions();
+        }
     }
 
     /**

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -556,7 +556,7 @@ public void testPrimaryFillsSeqNoGapsOnPromotion() throws Exception {
 
         // most of the time this is large enough that most of the time there will be at least one gap
         final int operations = 1024 - scaledRandomIntBetween(0, 1024);
-        final Result result = indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED), false);
+        final Result result = indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED));
 
         final int maxSeqNo = result.maxSeqNo;
         final boolean gap = result.gap;
@@ -1104,7 +1104,7 @@ public void testGlobalCheckpointSync() throws IOException {
     public void testRestoreLocalHistoryFromTranslogOnPromotion() throws IOException, InterruptedException {
         final IndexShard indexShard = newStartedShard(false);
         final int operations = 1024 - scaledRandomIntBetween(0, 1024);
-        indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED), true);
+        indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED));
 
         final long maxSeqNo = indexShard.seqNoStats().getMaxSeqNo();
         final long globalCheckpointOnReplica = randomLongBetween(UNASSIGNED_SEQ_NO, indexShard.getLocalCheckpoint());
@@ -1170,9 +1170,7 @@ public void testRollbackReplicaEngineOnPromotion() throws IOException, Interrupt
 
         // most of the time this is large enough that most of the time there will be at least one gap
         final int operations = 1024 - scaledRandomIntBetween(0, 1024);
-        // todo: all tests should run with allowUpdates=true, but this specific test sometimes fails during lucene commit when updates are
-        // added (seed = F37E9647ABE5928)
-        indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED), false);
+        indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED));
 
         final long globalCheckpointOnReplica = randomLongBetween(UNASSIGNED_SEQ_NO, indexShard.getLocalCheckpoint());
         indexShard.updateGlobalCheckpointOnReplica(globalCheckpointOnReplica, "test");
@@ -1215,7 +1213,7 @@ public void onFailure(final Exception e) {
         }
         assertThat(indexShard.getMaxSeqNoOfUpdatesOrDeletes(), equalTo(newMaxSeqNoOfUpdates));
         // ensure that after the local checkpoint throw back and indexing again, the local checkpoint advances
-        final Result result = indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(indexShard.getLocalCheckpoint()), false);
+        final Result result = indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(indexShard.getLocalCheckpoint()));
         assertThat(indexShard.getLocalCheckpoint(), equalTo((long) result.localCheckpoint));
         closeShard(indexShard, false);
     }
@@ -1473,6 +1471,12 @@ public String[] listAll() throws IOException {
                     return super.listAll();
                 }
             }
+
+            // temporary override until LUCENE-8735 is integrated
+            @Override
+            public Set<String> getPendingDeletions() throws IOException {
+                return in.getPendingDeletions();
+            }
         };
 
         try (Store store = createStore(shardId, new IndexSettings(metaData, Settings.EMPTY), directory)) {
@@ -3126,23 +3130,21 @@ class Result {
      * @param indexShard the shard
      * @param operations the number of operations
      * @param offset     the starting sequence number
-     * @param allowUpdates whether updates should be added.
      * @return a pair of the maximum sequence number and whether or not a gap was introduced
      * @throws IOException if an I/O exception occurs while indexing on the shard
      */
     private Result indexOnReplicaWithGaps(
             final IndexShard indexShard,
             final int operations,
-            final int offset,
-            boolean allowUpdates) throws IOException {
+            final int offset) throws IOException {
         int localCheckpoint = offset;
         int max = offset;
         boolean gap = false;
         Set<String> ids = new HashSet<>();
         for (int i = offset + 1; i < operations; i++) {
             if (!rarely() || i == operations - 1) { // last operation can't be a gap as it's not a gap anymore
                 final String id = ids.isEmpty() || randomBoolean() ? Integer.toString(i) : randomFrom(ids);
-                if (allowUpdates && ids.add(id) == false) { // this is an update
+                if (ids.add(id) == false) { // this is an update
                     indexShard.advanceMaxSeqNoOfUpdatesOrDeletes(i);
                 }
                 SourceToParse sourceToParse = SourceToParse.source(indexShard.shardId().getIndexName(), "_doc", id,
@@ -3557,7 +3559,7 @@ public void testSupplyTombstoneDoc() throws Exception {
 
     public void testResetEngine() throws Exception {
         IndexShard shard = newStartedShard(false);
-        indexOnReplicaWithGaps(shard, between(0, 1000), Math.toIntExact(shard.getLocalCheckpoint()), false);
+        indexOnReplicaWithGaps(shard, between(0, 1000), Math.toIntExact(shard.getLocalCheckpoint()));
         final long globalCheckpoint = randomLongBetween(shard.getGlobalCheckpoint(), shard.getLocalCheckpoint());
         shard.updateGlobalCheckpointOnReplica(globalCheckpoint, "test");
         Set<String> docBelowGlobalCheckpoint = getShardDocUIDs(shard).stream()
@@ -3597,7 +3599,7 @@ public void testResetEngine() throws Exception {
 
     public void testConcurrentAcquireAllReplicaOperationsPermitsWithPrimaryTermUpdate() throws Exception {
         final IndexShard replica = newStartedShard(false);
-        indexOnReplicaWithGaps(replica, between(0, 1000), Math.toIntExact(replica.getLocalCheckpoint()), false);
+        indexOnReplicaWithGaps(replica, between(0, 1000), Math.toIntExact(replica.getLocalCheckpoint()));
 
         final int nbTermUpdates = randomIntBetween(1, 5);
 

server/src/test/java/org/elasticsearch/index/store/ByteSizeCachingDirectoryTests.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.test.ESTestCase;
 
 import java.io.IOException;
+import java.util.Set;
 
 @LuceneTestCase.SuppressFileSystems("ExtrasFS")
 public class ByteSizeCachingDirectoryTests extends ESTestCase {
@@ -45,6 +46,12 @@ public long fileLength(String name) throws IOException {
             numFileLengthCalls++;
             return super.fileLength(name);
         }
+
+        // temporary override until LUCENE-8735 is integrated
+        @Override
+        public Set<String> getPendingDeletions() throws IOException {
+            return in.getPendingDeletions();
+        }
     }
 
     public void testBasics() throws IOException {

server/src/test/java/org/elasticsearch/index/store/StoreTests.java

@@ -43,9 +43,11 @@
 import org.apache.lucene.store.BaseDirectoryWrapper;
 import org.apache.lucene.store.ChecksumIndexInput;
 import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.FilterDirectory;
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
+import org.apache.lucene.store.NIOFSDirectory;
 import org.apache.lucene.store.RAMDirectory;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.TestUtil;
@@ -1143,4 +1145,16 @@ IndexWriter openWriter(Store store) throws IOException {
             .setOpenMode(IndexWriterConfig.OpenMode.APPEND);
         return new IndexWriter(store.directory(), iwc);
     }
+
+    public void testGetPendingFiles() throws IOException {
+        final ShardId shardId = new ShardId("index", "_na_", 1);
+        final String testfile = "testfile";
+        try (Store store = new Store(shardId, INDEX_SETTINGS, new NIOFSDirectory(createTempDir()), new DummyShardLock(shardId))) {
+            store.directory().createOutput(testfile, IOContext.DEFAULT).close();
+            try (IndexInput input = store.directory().openInput(testfile, IOContext.DEFAULT)) {
+                store.directory().deleteFile(testfile);
+                assertEquals(FilterDirectory.unwrap(store.directory()).getPendingDeletions(), store.directory().getPendingDeletions());
+            }
+        }
+    }
 }

test/framework/src/main/java/org/elasticsearch/test/store/MockFSDirectoryService.java

@@ -54,6 +54,7 @@
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Random;
+import java.util.Set;
 
 public class MockFSDirectoryService extends FsDirectoryService {
 
@@ -187,6 +188,12 @@ public synchronized void crash() throws IOException {
                 super.crash();
             }
         }
+
+        // temporary override until LUCENE-8735 is integrated
+        @Override
+        public Set<String> getPendingDeletions() throws IOException {
+            return in.getPendingDeletions();
+        }
     }
 
     final class CloseableDirectory implements Closeable {