diff --git a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeIndexingExecutionEngineTests.java b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeIndexingExecutionEngineTests.java index 50272163503c8..734ab86fd6ed2 100644 --- a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeIndexingExecutionEngineTests.java +++ b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeIndexingExecutionEngineTests.java @@ -288,6 +288,11 @@ public java.util.List li @Override public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } }; Map plugins = new HashMap<>(); @@ -356,6 +361,11 @@ public java.util.List li @Override public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } }; Map plugins = new HashMap<>(); @@ -427,6 +437,11 @@ public java.util.List li @Override public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } } private IndexSettings createIndexSettings(String primaryFormat) { diff --git a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeTestHelper.java b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeTestHelper.java index 028a3fd212723..1343872bc6743 100644 --- a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeTestHelper.java +++ b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeTestHelper.java @@ -263,5 +263,10 @@ public java.util.List li @Override public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } } } diff --git a/server/src/main/java/org/opensearch/index/engine/exec/CommitFileManager.java b/server/src/main/java/org/opensearch/index/engine/exec/CommitFileManager.java index 85e736a3b85ab..13ec17ca6c874 100644 --- a/server/src/main/java/org/opensearch/index/engine/exec/CommitFileManager.java +++ b/server/src/main/java/org/opensearch/index/engine/exec/CommitFileManager.java @@ -33,10 +33,12 @@ public interface CommitFileManager { /** * Returns true if the given file is managed by the commit mechanism * (e.g., segments_N, write.lock) and should not be treated as an orphan. + *

+ * Implementations MUST return true for all files they manage. Returning false + * for a commit-managed file will cause {@code IndexFileDeleter} to treat it + * as an orphan and delete it on startup, leading to data loss. * * @param fileName the file name to check */ - default boolean isCommitManagedFile(String fileName) { - return false; - } + boolean isCommitManagedFile(String fileName); } diff --git a/server/src/main/java/org/opensearch/index/engine/exec/coord/CatalogSnapshot.java b/server/src/main/java/org/opensearch/index/engine/exec/coord/CatalogSnapshot.java index ca7169cc535fd..cbb9dcddfd351 100644 --- a/server/src/main/java/org/opensearch/index/engine/exec/coord/CatalogSnapshot.java +++ b/server/src/main/java/org/opensearch/index/engine/exec/coord/CatalogSnapshot.java @@ -18,6 +18,9 @@ import java.io.IOException; import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -55,6 +58,12 @@ public abstract class CatalogSnapshot implements Writeable, Cloneable { private final AbstractRefCounted refCounter; + /** + * Lazily computed cache of file names grouped by format. Computed once on first access + * since segments are immutable after construction. + */ + private volatile Map> filesByFormatCache; + protected CatalogSnapshot(String name, long generation, long version) { this.generation = generation; this.version = version; @@ -177,6 +186,35 @@ int refCount() { */ public abstract String serializeToString() throws IOException; + /** + * Returns all file names grouped by data format name. The result is computed once + * and cached for the lifetime of this snapshot (segments are immutable after construction). + *

+ * Used by {@code IndexFileDeleter} for ref-count bookkeeping, avoiding repeated + * iteration over segments on every add/remove call. + * + * @return unmodifiable map of format name to unmodifiable set of file names + */ + public Map> getFilesByFormat() { + Map> cached = this.filesByFormatCache; + if (cached != null) { + return cached; + } + Map> result = new HashMap<>(); + for (Segment segment : getSegments()) { + for (Map.Entry entry : segment.dfGroupedSearchableFiles().entrySet()) { + result.computeIfAbsent(entry.getKey(), k -> new HashSet<>()).addAll(entry.getValue().files()); + } + } + Map> unmodifiable = new HashMap<>(); + for (Map.Entry> entry : result.entrySet()) { + unmodifiable.put(entry.getKey(), Collections.unmodifiableSet(entry.getValue())); + } + cached = Collections.unmodifiableMap(unmodifiable); + this.filesByFormatCache = cached; + return cached; + } + /** * Creates a clone without acquiring a reference count. * Used for Lucene compatibility where clone is required. diff --git a/server/src/main/java/org/opensearch/index/engine/exec/coord/IndexFileDeleter.java b/server/src/main/java/org/opensearch/index/engine/exec/coord/IndexFileDeleter.java index 7283fc4fdb280..fea9e87bd9791 100644 --- a/server/src/main/java/org/opensearch/index/engine/exec/coord/IndexFileDeleter.java +++ b/server/src/main/java/org/opensearch/index/engine/exec/coord/IndexFileDeleter.java @@ -17,10 +17,8 @@ import org.opensearch.index.engine.exec.FilesListener; import org.opensearch.index.shard.ShardPath; +import java.io.Closeable; import java.io.IOException; -import java.nio.file.DirectoryStream; -import java.nio.file.Files; -import java.nio.file.Path; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; @@ -51,7 +49,7 @@ * @opensearch.experimental */ @ExperimentalApi -public class IndexFileDeleter { +public class IndexFileDeleter implements Closeable { private static final Logger logger = LogManager.getLogger(IndexFileDeleter.class); @@ -62,6 +60,13 @@ public class IndexFileDeleter { private final List committedSnapshots; private final CommitFileManager commitFileManager; + /** + * Files that were dereferenced (ref count hit 0) but whose physical deletion + * failed. Retried on the next {@link #removeFileReferences}, {@link #revisitPolicy}, + * or {@link #retryPendingDeletes} call. + */ + private final Map> pendingDeletes; + public IndexFileDeleter( CatalogSnapshotDeletionPolicy deletionPolicy, Map fileDeleters, @@ -76,6 +81,7 @@ public IndexFileDeleter( this.fileRefCounts = new HashMap<>(); this.committedSnapshots = new ArrayList<>(); this.commitFileManager = commitFileManager; + this.pendingDeletes = new HashMap<>(); for (CatalogSnapshot cs : initialCommittedSnapshots) { if (cs.tryIncRef() == false) { @@ -85,6 +91,7 @@ public IndexFileDeleter( addFileReferences(cs); } + // No synchronization needed — `this` hasn't escaped the constructor yet. List toDelete = deletionPolicy.onInit(committedSnapshots); for (CatalogSnapshot old : toDelete) { committedSnapshots.remove(old); @@ -105,7 +112,7 @@ public IndexFileDeleter( * @return files whose ref count went from 0 to 1 (newly visible), grouped by format name */ public synchronized Map> addFileReferences(CatalogSnapshot snapshot) throws IOException { - Map> segregated = segregateFilesByFormat(snapshot); + Map> segregated = snapshot.getFilesByFormat(); Map> newFiles = new HashMap<>(); for (Map.Entry> entry : segregated.entrySet()) { @@ -117,6 +124,14 @@ public synchronized Map> addFileReferences(CatalogSna AtomicInteger refCount = formatRefs.computeIfAbsent(file, k -> new AtomicInteger(0)); if (refCount.incrementAndGet() == 1) { added.add(file); + // File is live again — remove from pending deletes if present + Set pending = pendingDeletes.get(formatName); + if (pending != null) { + pending.remove(file); + if (pending.isEmpty()) { + pendingDeletes.remove(formatName); + } + } } } if (added.isEmpty() == false) { @@ -130,19 +145,27 @@ public synchronized Map> addFileReferences(CatalogSna /** * Called when a CatalogSnapshot's refCount reaches 0. * Decrements ref counts for its files and deletes files that hit 0. + *

+ * Ref count bookkeeping is done under the lock; actual I/O (commit deletion, + * file deletion, listener notification) is performed outside the lock to avoid + * blocking concurrent {@link #addFileReferences} and {@link #onCommit} calls. */ - public synchronized void removeFileReferences(CatalogSnapshot snapshot) throws IOException { - Map> filesToDelete = decRefFiles(snapshot); + public void removeFileReferences(CatalogSnapshot snapshot) throws IOException { + Map> filesToDelete; + synchronized (this) { + filesToDelete = decRefFiles(snapshot); + } // Delete the commit point (segments_N) BEFORE deleting data files, // because deleteCommit may call DirectoryReader.listCommits() which // needs to read segment files that are about to be deleted. if (commitFileManager != null) { commitFileManager.deleteCommit(snapshot); - } else {} + } if (filesToDelete.isEmpty() == false) { - deleteFiles(filesToDelete); - notifyFilesDeleted(filesToDelete); + executeDeletesWithRetry(filesToDelete); } + // Also retry any previously failed deletes + retryPendingDeletes(); } // ---- Commit path ---- @@ -154,16 +177,20 @@ public synchronized void removeFileReferences(CatalogSnapshot snapshot) throws I * Snapshots the policy wants to delete get their commit ref decRef'd, * which may trigger removeFileReferences when their refCount reaches 0. */ - public synchronized void onCommit(CatalogSnapshot committedSnapshot) throws IOException { - committedSnapshots.add(committedSnapshot); - - List toDelete = deletionPolicy.onCommit(committedSnapshots); + public void onCommit(CatalogSnapshot committedSnapshot) throws IOException { + List toDelete; + synchronized (this) { + committedSnapshots.add(committedSnapshot); + toDelete = deletionPolicy.onCommit(committedSnapshots); + for (CatalogSnapshot old : toDelete) { + committedSnapshots.remove(old); + } + } for (CatalogSnapshot old : toDelete) { - committedSnapshots.remove(old); if (old.decRef()) { removeFileReferences(old); - } else {} + } } } @@ -174,27 +201,106 @@ public synchronized void onCommit(CatalogSnapshot committedSnapshot) throws IOEx * Used after releasing a snapshot hold — the policy may now allow * deletion of commits it previously protected. */ - public synchronized void revisitPolicy() throws IOException { - List toDelete = deletionPolicy.onCommit(committedSnapshots); + public void revisitPolicy() throws IOException { + List toDelete; + synchronized (this) { + toDelete = deletionPolicy.onCommit(committedSnapshots); + for (CatalogSnapshot old : toDelete) { + committedSnapshots.remove(old); + } + } for (CatalogSnapshot old : toDelete) { - committedSnapshots.remove(old); if (old.decRef()) { removeFileReferences(old); } } } - // ---- Listener notification ---- - - private void deleteFiles(Map> filesByFormat) throws IOException { - for (Map.Entry> entry : filesByFormat.entrySet()) { - FileDeleter deleter = fileDeleters.get(entry.getKey()); - if (deleter != null) { - deleter.deleteFiles(Map.of(entry.getKey(), entry.getValue())); + /** + * Retries deletion of files that failed to delete on a previous attempt. + * Can be called periodically or after operations that may have released + * filesystem locks. + */ + public void retryPendingDeletes() throws IOException { + Map> snapshot; + synchronized (this) { + if (pendingDeletes.isEmpty()) { + return; + } + // Take a snapshot of pending deletes to process outside the lock + snapshot = new HashMap<>(); + for (Map.Entry> entry : pendingDeletes.entrySet()) { + snapshot.put(entry.getKey(), new HashSet<>(entry.getValue())); } } + for (Map.Entry> entry : snapshot.entrySet()) { + String formatName = entry.getKey(); + Set files = entry.getValue(); + FileDeleter deleter = fileDeleters.get(formatName); + if (deleter == null) { + continue; + } + Set stillFailed = new HashSet<>(); + for (String file : files) { + // Re-check: skip if file was re-referenced since we snapshotted pendingDeletes + boolean isLive; + synchronized (this) { + Map formatRefs = fileRefCounts.get(formatName); + isLive = formatRefs != null && formatRefs.containsKey(file); + if (isLive) { + // Remove from pending — addFileReferences should have done this, + // but guard against edge cases + Set pending = pendingDeletes.get(formatName); + if (pending != null) { + pending.remove(file); + if (pending.isEmpty()) { + pendingDeletes.remove(formatName); + } + } + } + } + if (isLive) { + continue; + } + try { + deleter.deleteFiles(Map.of(formatName, List.of(file))); + } catch (IOException e) { + logger.warn("Retry delete failed for [{}] in format [{}]", file, formatName, e); + stillFailed.add(file); + } + } + synchronized (this) { + Set currentPending = pendingDeletes.get(formatName); + if (currentPending != null) { + // Remove successfully deleted files + currentPending.retainAll(stillFailed); + if (currentPending.isEmpty()) { + pendingDeletes.remove(formatName); + } + } + } + } + // Notify listeners for files that were successfully deleted in retry + Map> deleted = new HashMap<>(); + for (Map.Entry> entry : snapshot.entrySet()) { + synchronized (this) { + Set currentPending = pendingDeletes.get(entry.getKey()); + Set succeeded = new HashSet<>(entry.getValue()); + if (currentPending != null) { + succeeded.removeAll(currentPending); + } + if (succeeded.isEmpty() == false) { + deleted.put(entry.getKey(), succeeded); + } + } + } + if (deleted.isEmpty() == false) { + notifyFilesDeleted(deleted); + } } + // ---- Listener notification (outside lock) ---- + private void notifyFilesAdded(Map> newFilesByFormat) throws IOException { for (Map.Entry> entry : newFilesByFormat.entrySet()) { FilesListener listener = filesListeners.get(entry.getKey()); @@ -213,10 +319,59 @@ private void notifyFilesDeleted(Map> deletedFilesByFo } } + // ---- Deletion with pending-delete tracking ---- + + /** + * Attempts to delete files. Before deleting, re-checks under the lock that each file + * is still unreferenced — a concurrent {@link #addFileReferences} may have re-referenced + * a file between {@link #decRefFiles} and this call. Files that fail to delete are added + * to {@link #pendingDeletes} for retry. Successfully deleted files trigger listener notification. + */ + private void executeDeletesWithRetry(Map> filesByFormat) throws IOException { + // Filter out files that were re-referenced since decRefFiles ran + Map> safeToDelete = new HashMap<>(); + synchronized (this) { + for (Map.Entry> entry : filesByFormat.entrySet()) { + String formatName = entry.getKey(); + Map formatRefs = fileRefCounts.get(formatName); + Collection stillDead = new HashSet<>(); + for (String file : entry.getValue()) { + if (formatRefs == null || formatRefs.containsKey(file) == false) { + stillDead.add(file); + } + } + if (stillDead.isEmpty() == false) { + safeToDelete.put(formatName, stillDead); + } + } + } + Map> successfullyDeleted = new HashMap<>(); + for (Map.Entry> entry : safeToDelete.entrySet()) { + String formatName = entry.getKey(); + Collection files = entry.getValue(); + FileDeleter deleter = fileDeleters.get(formatName); + if (deleter != null) { + try { + deleter.deleteFiles(Map.of(formatName, files)); + successfullyDeleted.put(formatName, files); + } catch (IOException e) { + logger.warn("Failed to delete files for format [{}], adding to pending deletes", formatName, e); + synchronized (this) { + pendingDeletes.computeIfAbsent(formatName, k -> new HashSet<>()).addAll(files); + } + } + } + } + if (successfullyDeleted.isEmpty() == false) { + notifyFilesDeleted(successfullyDeleted); + } + } + // ---- Internal ---- private Map> decRefFiles(CatalogSnapshot snapshot) { - Map> segregated = segregateFilesByFormat(snapshot); + assert Thread.holdsLock(this); + Map> segregated = snapshot.getFilesByFormat(); Map> filesToDelete = new HashMap<>(); for (Map.Entry> entry : segregated.entrySet()) { @@ -240,45 +395,56 @@ private Map> decRefFiles(CatalogSnapshot snapshot) { return filesToDelete; } - private Map> segregateFilesByFormat(CatalogSnapshot snapshot) { - Map> result = new HashMap<>(); - for (var segment : snapshot.getSegments()) { - for (var entry : segment.dfGroupedSearchableFiles().entrySet()) { - result.computeIfAbsent(entry.getKey(), k -> new HashSet<>()).addAll(entry.getValue().files()); - } + /** + * Scans format directories for files not tracked by any snapshot and deletes them. + * Called once during construction to clean up after a crash between ref-count + * decrement and physical deletion. + */ + private void deleteOrphanedFiles(ShardPath shardPath) throws IOException { + Map> knownFilesByFormat = new HashMap<>(); + for (Map.Entry> entry : fileRefCounts.entrySet()) { + knownFilesByFormat.put(entry.getKey(), new HashSet<>(entry.getValue().keySet())); + } + Map> orphans = OrphanFileScanner.findOrphans(shardPath, knownFilesByFormat, commitFileManager); + if (orphans.isEmpty() == false) { + executeDeletesWithRetry(orphans); } - return result; } - private void deleteOrphanedFiles(ShardPath shardPath) throws IOException { - if (shardPath == null) { - return; + /** + * Returns true if there are files awaiting deletion retry. + * Visible for testing. + */ + public synchronized boolean hasPendingDeletes() { + return pendingDeletes.isEmpty() == false; + } + + /** + * Returns a snapshot of the current pending deletes map. + * Visible for testing. + */ + synchronized Map> getPendingDeletes() { + Map> copy = new HashMap<>(); + for (Map.Entry> entry : pendingDeletes.entrySet()) { + copy.put(entry.getKey(), new HashSet<>(entry.getValue())); } - Map> orphansByFormat = new HashMap<>(); - for (Map.Entry> entry : fileRefCounts.entrySet()) { - String formatName = entry.getKey(); - Set knownFiles = entry.getValue().keySet(); - Path formatDir = "lucene".equals(formatName) ? shardPath.resolveIndex() : shardPath.getDataPath().resolve(formatName); - if (Files.exists(formatDir) == false) { - continue; - } - Collection orphans = new HashSet<>(); - try (DirectoryStream stream = Files.newDirectoryStream(formatDir)) { - for (Path file : stream) { - String fileName = file.getFileName().toString(); - if (Files.isRegularFile(file) - && knownFiles.contains(fileName) == false - && (commitFileManager == null || commitFileManager.isCommitManagedFile(fileName) == false)) { - orphans.add(fileName); - } - } - } - if (orphans.isEmpty() == false) { - orphansByFormat.put(formatName, orphans); - } + return copy; + } + + /** + * Releases all committed snapshot refs held by this deleter and drains pending deletes. + * Should be called during engine shutdown to prevent ref leaks. + */ + @Override + public void close() throws IOException { + List snapshotsToRelease; + synchronized (this) { + snapshotsToRelease = new ArrayList<>(committedSnapshots); + committedSnapshots.clear(); + pendingDeletes.clear(); } - if (orphansByFormat.isEmpty() == false) { - deleteFiles(orphansByFormat); + for (CatalogSnapshot cs : snapshotsToRelease) { + cs.decRef(); } } diff --git a/server/src/main/java/org/opensearch/index/engine/exec/coord/OrphanFileScanner.java b/server/src/main/java/org/opensearch/index/engine/exec/coord/OrphanFileScanner.java new file mode 100644 index 0000000000000..257d992324349 --- /dev/null +++ b/server/src/main/java/org/opensearch/index/engine/exec/coord/OrphanFileScanner.java @@ -0,0 +1,79 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.index.engine.exec.coord; + +import org.opensearch.index.engine.exec.CommitFileManager; +import org.opensearch.index.shard.ShardPath; + +import java.io.IOException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Scans format directories on disk and identifies files that are not referenced + * by any known snapshot and not managed by the commit mechanism. + *

+ * Stateless utility — does not hold any mutable state. Called once during + * {@link IndexFileDeleter} construction to clean up files left behind by + * a crash between ref-count decrement and physical deletion. + */ +final class OrphanFileScanner { + + private OrphanFileScanner() {} + + /** + * Finds orphaned files across all known format directories. + * + * @param shardPath the shard's data path (null skips the scan) + * @param knownFilesByFormat map of format name to the set of file names currently tracked + * @param commitFileManager optional manager that identifies commit-owned files (may be null) + * @return map of format name to orphaned file names; empty if nothing to clean up + */ + static Map> findOrphans( + ShardPath shardPath, + Map> knownFilesByFormat, + CommitFileManager commitFileManager + ) throws IOException { + if (shardPath == null) { + return Map.of(); + } + Map> orphansByFormat = new HashMap<>(); + for (Map.Entry> entry : knownFilesByFormat.entrySet()) { + String formatName = entry.getKey(); + Set knownFiles = entry.getValue(); + Path formatDir = "lucene".equals(formatName) + ? shardPath.resolveIndex() + : shardPath.getDataPath().resolve(formatName); + if (Files.exists(formatDir) == false) { + continue; + } + Collection orphans = new HashSet<>(); + try (DirectoryStream stream = Files.newDirectoryStream(formatDir)) { + for (Path file : stream) { + String fileName = file.getFileName().toString(); + if (Files.isRegularFile(file) + && knownFiles.contains(fileName) == false + && (commitFileManager == null || commitFileManager.isCommitManagedFile(fileName) == false)) { + orphans.add(fileName); + } + } + } + if (orphans.isEmpty() == false) { + orphansByFormat.put(formatName, orphans); + } + } + return orphansByFormat; + } +} diff --git a/server/src/test/java/org/opensearch/index/engine/DataFormatAwareEngineCommitterTests.java b/server/src/test/java/org/opensearch/index/engine/DataFormatAwareEngineCommitterTests.java index afe79d1e7dca8..22a502390f7d5 100644 --- a/server/src/test/java/org/opensearch/index/engine/DataFormatAwareEngineCommitterTests.java +++ b/server/src/test/java/org/opensearch/index/engine/DataFormatAwareEngineCommitterTests.java @@ -55,6 +55,11 @@ public java.util.List li @Override public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } }; engine.setCommitter(committer); assertSame(committer, engine.getCommitter()); diff --git a/server/src/test/java/org/opensearch/index/engine/exec/commit/CommitterTests.java b/server/src/test/java/org/opensearch/index/engine/exec/commit/CommitterTests.java index 4247831d375b9..4af0e2b148c85 100644 --- a/server/src/test/java/org/opensearch/index/engine/exec/commit/CommitterTests.java +++ b/server/src/test/java/org/opensearch/index/engine/exec/commit/CommitterTests.java @@ -51,6 +51,11 @@ public java.util.List li @Override public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } }; } @@ -91,6 +96,11 @@ public java.util.List li @Override public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } }; committer.close(); assertTrue("close() should have been called", closed.get()); @@ -129,6 +139,11 @@ public java.util.List li @Override public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } }; committer.commit(Map.of()); assertTrue("commit() should have been called", committed.get()); diff --git a/server/src/test/java/org/opensearch/index/engine/exec/coord/IndexFileDeleterTests.java b/server/src/test/java/org/opensearch/index/engine/exec/coord/IndexFileDeleterTests.java index 706980bab7295..f605f100cfcc4 100644 --- a/server/src/test/java/org/opensearch/index/engine/exec/coord/IndexFileDeleterTests.java +++ b/server/src/test/java/org/opensearch/index/engine/exec/coord/IndexFileDeleterTests.java @@ -11,6 +11,7 @@ import org.opensearch.core.index.shard.ShardId; import org.opensearch.index.engine.exec.CatalogSnapshotDeletionPolicy; import org.opensearch.index.engine.exec.CombinedCatalogSnapshotDeletionPolicy; +import org.opensearch.index.engine.exec.CommitFileManager; import org.opensearch.index.engine.exec.FileDeleter; import org.opensearch.index.engine.exec.Segment; import org.opensearch.index.engine.exec.WriterFileSet; @@ -23,6 +24,7 @@ import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; @@ -381,4 +383,367 @@ public void testDeleteOrphanedFilesSkipsMissingDirectory() throws IOException { ); assertTrue(tracker.deletedFiles.isEmpty()); } + + // ---- Fix #3: Partial deletion failure tracking via pendingDeletes ---- + + /** + * A FileDeleter that fails the first N calls, then succeeds on subsequent calls. + * Simulates transient I/O failure (e.g., remote store timeout). + */ + static class FailNTimesThenSucceedDeleter implements FileDeleter { + final Set deletedFiles = new HashSet<>(); + private int failuresRemaining; + + FailNTimesThenSucceedDeleter(int failCount) { + this.failuresRemaining = failCount; + } + + @Override + public void deleteFiles(Map> filesToDelete) throws IOException { + if (failuresRemaining > 0) { + failuresRemaining--; + throw new IOException("Simulated transient I/O failure"); + } + for (Collection files : filesToDelete.values()) { + deletedFiles.addAll(files); + } + } + } + + /** + * A FileDeleter that always throws IOException. Simulates persistent I/O failure. + */ + static class AlwaysFailingDeleter implements FileDeleter { + int deleteAttempts = 0; + + @Override + public void deleteFiles(Map> filesToDelete) throws IOException { + deleteAttempts++; + throw new IOException("Persistent I/O failure"); + } + } + + public void testPartialDeleteFailureTracksPendingDeletes() throws IOException { + AlwaysFailingDeleter failingDeleter = new AlwaysFailingDeleter(); + CatalogSnapshot cs1 = snapshot(1, List.of(segment(0, "parquet", "a.parquet", "b.parquet")), commitUserData(100, 100, "uuid")); + + IndexFileDeleter deleter = new IndexFileDeleter( + CatalogSnapshotDeletionPolicy.KEEP_LATEST_ONLY, + Map.of("parquet", failingDeleter), + Map.of(), + List.of(cs1), + null, + null + ); + + // Add cs2 with different files, then remove cs1 + CatalogSnapshot cs2 = snapshot(2, List.of(segment(1, "parquet", "c.parquet")), commitUserData(200, 200, "uuid")); + deleter.addFileReferences(cs2); + deleter.removeFileReferences(cs1); + + // Files should be in pending deletes since the deleter always fails + assertTrue("Should have pending deletes after failed deletion", deleter.hasPendingDeletes()); + Map> pending = deleter.getPendingDeletes(); + assertTrue("a.parquet should be pending", pending.get("parquet").contains("a.parquet")); + assertTrue("b.parquet should be pending", pending.get("parquet").contains("b.parquet")); + } + + public void testPendingDeletesRetriedOnNextRemoveFileReferences() throws IOException { + // Fail twice: once during executeDeletesWithRetry, once during the automatic + // retryPendingDeletes inside removeFileReferences. Third call succeeds. + FailNTimesThenSucceedDeleter deleter1 = new FailNTimesThenSucceedDeleter(2); + CatalogSnapshot cs1 = snapshot(1, List.of(segment(0, "parquet", "a.parquet")), commitUserData(100, 100, "uuid")); + + IndexFileDeleter deleter = new IndexFileDeleter( + CatalogSnapshotDeletionPolicy.KEEP_LATEST_ONLY, + Map.of("parquet", deleter1), + Map.of(), + List.of(cs1), + null, + null + ); + + // cs2 and cs3 with different files + CatalogSnapshot cs2 = snapshot(2, List.of(segment(1, "parquet", "b.parquet")), commitUserData(200, 200, "uuid")); + deleter.addFileReferences(cs2); + CatalogSnapshot cs3 = snapshot(3, List.of(segment(2, "parquet", "c.parquet")), commitUserData(300, 300, "uuid")); + deleter.addFileReferences(cs3); + + // First removal: executeDeletesWithRetry fails (call 1), retryPendingDeletes fails (call 2) + deleter.removeFileReferences(cs1); + assertTrue("Should have pending deletes after both attempts failed", deleter.hasPendingDeletes()); + + // Second removal: deleter now succeeds (call 3+), retrying a.parquet and deleting b.parquet + deleter.removeFileReferences(cs2); + assertTrue("a.parquet should be deleted on retry", deleter1.deletedFiles.contains("a.parquet")); + assertTrue("b.parquet should be deleted", deleter1.deletedFiles.contains("b.parquet")); + assertFalse("No more pending deletes", deleter.hasPendingDeletes()); + } + + public void testPendingDeletesClearedWhenFileReReferencedByNewSnapshot() throws IOException { + AlwaysFailingDeleter failingDeleter = new AlwaysFailingDeleter(); + CatalogSnapshot cs1 = snapshot(1, List.of(segment(0, "parquet", "shared.parquet")), commitUserData(100, 100, "uuid")); + + IndexFileDeleter deleter = new IndexFileDeleter( + CatalogSnapshotDeletionPolicy.KEEP_LATEST_ONLY, + Map.of("parquet", failingDeleter), + Map.of(), + List.of(cs1), + null, + null + ); + + // cs2 does NOT contain shared.parquet + CatalogSnapshot cs2 = snapshot(2, List.of(segment(1, "parquet", "other.parquet")), commitUserData(200, 200, "uuid")); + deleter.addFileReferences(cs2); + deleter.removeFileReferences(cs1); + + // shared.parquet should be pending (deletion failed) + assertTrue(deleter.hasPendingDeletes()); + assertTrue(deleter.getPendingDeletes().get("parquet").contains("shared.parquet")); + + // cs3 re-introduces shared.parquet — it's live again + CatalogSnapshot cs3 = snapshot(3, List.of(segment(2, "parquet", "shared.parquet")), commitUserData(300, 300, "uuid")); + deleter.addFileReferences(cs3); + + // shared.parquet should be removed from pending deletes since it's referenced again + Map> pending = deleter.getPendingDeletes(); + boolean sharedStillPending = pending.containsKey("parquet") && pending.get("parquet").contains("shared.parquet"); + assertFalse("Re-referenced file should be removed from pending deletes", sharedStillPending); + } + + public void testRetryPendingDeletesExplicitCall() throws IOException { + // Fail twice: once during executeDeletesWithRetry, once during the automatic + // retryPendingDeletes inside removeFileReferences. Third call (explicit retry) succeeds. + FailNTimesThenSucceedDeleter failTwice = new FailNTimesThenSucceedDeleter(2); + CatalogSnapshot cs1 = snapshot(1, List.of(segment(0, "parquet", "a.parquet")), commitUserData(100, 100, "uuid")); + + IndexFileDeleter deleter = new IndexFileDeleter( + CatalogSnapshotDeletionPolicy.KEEP_LATEST_ONLY, + Map.of("parquet", failTwice), + Map.of(), + List.of(cs1), + null, + null + ); + + CatalogSnapshot cs2 = snapshot(2, List.of(segment(1, "parquet", "b.parquet")), commitUserData(200, 200, "uuid")); + deleter.addFileReferences(cs2); + deleter.removeFileReferences(cs1); + + // Both internal attempts failed, file is still pending + assertTrue("Should have pending deletes", deleter.hasPendingDeletes()); + + // Explicit retry — deleter now succeeds (third call) + deleter.retryPendingDeletes(); + assertTrue("a.parquet should be deleted on explicit retry", failTwice.deletedFiles.contains("a.parquet")); + assertFalse("Pending deletes should be empty after successful retry", deleter.hasPendingDeletes()); + } + + public void testPersistentFailureKeepsFilesPending() throws IOException { + AlwaysFailingDeleter alwaysFails = new AlwaysFailingDeleter(); + CatalogSnapshot cs1 = snapshot(1, List.of(segment(0, "parquet", "doomed.parquet")), commitUserData(100, 100, "uuid")); + + IndexFileDeleter deleter = new IndexFileDeleter( + CatalogSnapshotDeletionPolicy.KEEP_LATEST_ONLY, + Map.of("parquet", alwaysFails), + Map.of(), + List.of(cs1), + null, + null + ); + + CatalogSnapshot cs2 = snapshot(2, List.of(segment(1, "parquet", "ok.parquet")), commitUserData(200, 200, "uuid")); + deleter.addFileReferences(cs2); + deleter.removeFileReferences(cs1); + + assertTrue(deleter.hasPendingDeletes()); + int attemptsAfterRemove = alwaysFails.deleteAttempts; + + // Retry — still fails + deleter.retryPendingDeletes(); + assertTrue("File should remain pending after persistent failure", deleter.hasPendingDeletes()); + assertTrue("doomed.parquet still pending", deleter.getPendingDeletes().get("parquet").contains("doomed.parquet")); + assertTrue("Should have attempted more deletes", alwaysFails.deleteAttempts > attemptsAfterRemove); + } + + // ---- Fix #1: I/O outside the synchronized block ---- + + /** + * A FileDeleter that records which thread performed the deletion and whether + * the IndexFileDeleter monitor was held at that time. + */ + static class LockProbeDeleter implements FileDeleter { + final Set deletedFiles = new HashSet<>(); + final List lockHeldDuringDelete = new ArrayList<>(); + private final Object monitorToProbe; + + LockProbeDeleter(Object monitorToProbe) { + this.monitorToProbe = monitorToProbe; + } + + @Override + public void deleteFiles(Map> filesToDelete) { + lockHeldDuringDelete.add(Thread.holdsLock(monitorToProbe)); + for (Collection files : filesToDelete.values()) { + deletedFiles.addAll(files); + } + } + } + + public void testDeleteFilesExecutedOutsideSynchronizedBlock() throws IOException { + AtomicLong globalCP = new AtomicLong(100); + + CombinedCatalogSnapshotDeletionPolicy policy = new CombinedCatalogSnapshotDeletionPolicy( + logger, + new DefaultTranslogDeletionPolicy(-1, -1, 0), + globalCP::get + ); + + CatalogSnapshot cs1 = snapshot(1, List.of(segment(0, "parquet", "old.parquet")), commitUserData(100, 100, "uuid")); + // Create deleter first, then set up the probe + IndexFileDeleter deleter = new IndexFileDeleter(policy, Map.of("parquet", new TrackingFileDeleter()), Map.of(), List.of(cs1), null, null); + + // Now create a new deleter with the lock probe, using the deleter instance as the monitor + LockProbeDeleter probe = new LockProbeDeleter(deleter); + + // We need a fresh deleter with the probe. Rebuild. + CatalogSnapshot cs1b = snapshot(1, List.of(segment(0, "parquet", "old.parquet")), commitUserData(100, 100, "uuid")); + IndexFileDeleter deleterWithProbe = new IndexFileDeleter( + policy, + Map.of("parquet", probe), + Map.of(), + List.of(cs1b), + null, + null + ); + + CatalogSnapshot cs2 = snapshot(2, List.of(segment(1, "parquet", "new.parquet")), commitUserData(200, 200, "uuid")); + deleterWithProbe.addFileReferences(cs2); + cs1b.decRef(); + globalCP.set(200); + deleterWithProbe.onCommit(cs2); + + // The probe should have recorded that the lock was NOT held during deleteFiles + assertTrue("old.parquet should have been deleted", probe.deletedFiles.contains("old.parquet")); + assertFalse("Lock should not be held during file deletion I/O", probe.lockHeldDuringDelete.isEmpty()); + for (Boolean held : probe.lockHeldDuringDelete) { + assertFalse("IndexFileDeleter monitor should NOT be held during deleteFiles call", held); + } + } + + public void testRemoveFileReferencesDoesNotHoldLockDuringIO() throws IOException { + CatalogSnapshot cs1 = snapshot(1, List.of(segment(0, "parquet", "a.parquet")), commitUserData(100, 100, "uuid")); + + // Placeholder deleter for construction + IndexFileDeleter deleter = new IndexFileDeleter( + CatalogSnapshotDeletionPolicy.KEEP_LATEST_ONLY, + Map.of("parquet", new TrackingFileDeleter()), + Map.of(), + List.of(cs1), + null, + null + ); + + // Rebuild with lock probe + LockProbeDeleter probe = new LockProbeDeleter(deleter); + CatalogSnapshot cs1b = snapshot(1, List.of(segment(0, "parquet", "a.parquet")), commitUserData(100, 100, "uuid")); + IndexFileDeleter deleterWithProbe = new IndexFileDeleter( + CatalogSnapshotDeletionPolicy.KEEP_LATEST_ONLY, + Map.of("parquet", probe), + Map.of(), + List.of(cs1b), + null, + null + ); + + CatalogSnapshot cs2 = snapshot(2, List.of(segment(1, "parquet", "b.parquet")), commitUserData(200, 200, "uuid")); + deleterWithProbe.addFileReferences(cs2); + deleterWithProbe.removeFileReferences(cs1b); + + assertTrue("a.parquet should be deleted", probe.deletedFiles.contains("a.parquet")); + for (Boolean held : probe.lockHeldDuringDelete) { + assertFalse("Monitor should NOT be held during removeFileReferences I/O", held); + } + } + + // ---- Fix #14: isCommitManagedFile is now abstract ---- + + public void testOrphanScanRespectsCommitManagedFiles() throws IOException { + TrackingFileDeleter tracker = new TrackingFileDeleter(); + + Path tempDir = createTempDir(); + ShardId shardId = new ShardId("test", "test", 0); + Path shardDir = tempDir.resolve(shardId.getIndex().getUUID()).resolve(String.valueOf(shardId.id())); + Files.createDirectories(shardDir); + + Path parquetDir = shardDir.resolve("parquet"); + Files.createDirectories(parquetDir); + Files.createFile(parquetDir.resolve("known.parquet")); + Files.createFile(parquetDir.resolve("orphan.parquet")); + Files.createFile(parquetDir.resolve("segments_1")); + + ShardPath shardPath = new ShardPath(false, shardDir, shardDir, shardId); + + // CommitFileManager that protects segments_* files + CommitFileManager commitMgr = new CommitFileManager() { + @Override + public void deleteCommit(CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return fileName.startsWith("segments_"); + } + }; + + CatalogSnapshot cs1 = snapshot(1, List.of(segment(0, "parquet", "known.parquet")), commitUserData(100, 100, "uuid")); + + IndexFileDeleter deleter = new IndexFileDeleter( + CatalogSnapshotDeletionPolicy.KEEP_LATEST_ONLY, + Map.of("parquet", tracker), + Map.of(), + List.of(cs1), + shardPath, + commitMgr + ); + + // orphan.parquet should be deleted (not referenced, not commit-managed) + assertTrue("orphan.parquet should be deleted", tracker.hasDeletedFile("orphan.parquet")); + // segments_1 should NOT be deleted (commit-managed) + assertFalse("segments_1 should be protected by CommitFileManager", tracker.hasDeletedFile("segments_1")); + // known.parquet should NOT be deleted (referenced) + assertFalse("known.parquet should not be deleted", tracker.hasDeletedFile("known.parquet")); + } + + public void testOrphanScanWithNullCommitFileManagerDeletesEverythingUnreferenced() throws IOException { + TrackingFileDeleter tracker = new TrackingFileDeleter(); + + Path tempDir = createTempDir(); + ShardId shardId = new ShardId("test", "test", 0); + Path shardDir = tempDir.resolve(shardId.getIndex().getUUID()).resolve(String.valueOf(shardId.id())); + Files.createDirectories(shardDir); + + Path parquetDir = shardDir.resolve("parquet"); + Files.createDirectories(parquetDir); + Files.createFile(parquetDir.resolve("known.parquet")); + Files.createFile(parquetDir.resolve("segments_1")); + + ShardPath shardPath = new ShardPath(false, shardDir, shardDir, shardId); + + CatalogSnapshot cs1 = snapshot(1, List.of(segment(0, "parquet", "known.parquet")), commitUserData(100, 100, "uuid")); + + // null commitFileManager — no protection for commit files + IndexFileDeleter deleter = new IndexFileDeleter( + CatalogSnapshotDeletionPolicy.KEEP_LATEST_ONLY, + Map.of("parquet", tracker), + Map.of(), + List.of(cs1), + shardPath, + null + ); + + // With null CommitFileManager, segments_1 is treated as an orphan + assertTrue("segments_1 should be deleted when no CommitFileManager protects it", tracker.hasDeletedFile("segments_1")); + assertFalse("known.parquet should not be deleted", tracker.hasDeletedFile("known.parquet")); + } } diff --git a/server/src/test/java/org/opensearch/index/engine/exec/coord/SafeBootstrapCommitterTests.java b/server/src/test/java/org/opensearch/index/engine/exec/coord/SafeBootstrapCommitterTests.java index 6a145a666a369..f8653c8cbf8b2 100644 --- a/server/src/test/java/org/opensearch/index/engine/exec/coord/SafeBootstrapCommitterTests.java +++ b/server/src/test/java/org/opensearch/index/engine/exec/coord/SafeBootstrapCommitterTests.java @@ -86,6 +86,11 @@ public List listCommittedSnapshots() { @Override public void deleteCommit(CatalogSnapshot snapshot) {} + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } + @Override public void close() {} }