Move trimming unsafe commits from engine ctor to store (#29260)

As follow up to #28245 , this PR removes the logic for selecting the
right start commit from the Engine constructor in favor of explicitly
trimming them in the Store, before the engine is opened. This makes the
constructor in engine follow standard Lucene semantics and use the last
commit.

Relates #28245
Relates #29156

docs/reference/indices/flush.asciidoc

@@ -93,12 +93,12 @@ which returns something similar to:
                {
                  "commit" : {
                    "id" : "3M3zkw2GHMo2Y4h4/KFKCg==",
-                   "generation" : 4,
+                   "generation" : 3,
                    "user_data" : {
                      "translog_uuid" : "hnOG3xFcTDeoI_kvvvOdNA",
                      "history_uuid" : "XP7KDJGiS1a2fHYiFL5TXQ",
                      "local_checkpoint" : "-1",
-                     "translog_generation" : "3",
+                     "translog_generation" : "2",
                      "max_seq_no" : "-1",
                      "sync_id" : "AVvFY-071siAOuFGEO9P", <1>
                      "max_unsafe_auto_id_timestamp" : "-1"

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

@@ -47,60 +47,27 @@ public final class CombinedDeletionPolicy extends IndexDeletionPolicy {
     private final Logger logger;
     private final TranslogDeletionPolicy translogDeletionPolicy;
     private final LongSupplier globalCheckpointSupplier;
-    private final IndexCommit startingCommit;
     private final ObjectIntHashMap<IndexCommit> snapshottedCommits; // Number of snapshots held against each commit point.
     private volatile IndexCommit safeCommit; // the most recent safe commit point - its max_seqno at most the persisted global checkpoint.
     private volatile IndexCommit lastCommit; // the most recent commit point
 
-    CombinedDeletionPolicy(Logger logger, TranslogDeletionPolicy translogDeletionPolicy,
-                           LongSupplier globalCheckpointSupplier, IndexCommit startingCommit) {
+    CombinedDeletionPolicy(Logger logger, TranslogDeletionPolicy translogDeletionPolicy, LongSupplier globalCheckpointSupplier) {
         this.logger = logger;
         this.translogDeletionPolicy = translogDeletionPolicy;
         this.globalCheckpointSupplier = globalCheckpointSupplier;
-        this.startingCommit = startingCommit;
         this.snapshottedCommits = new ObjectIntHashMap<>();
     }
 
     @Override
     public synchronized void onInit(List<? extends IndexCommit> commits) throws IOException {
         assert commits.isEmpty() == false : "index is opened, but we have no commits";
-        assert startingCommit != null && commits.contains(startingCommit) : "Starting commit not in the existing commit list; "
-            + "startingCommit [" + startingCommit + "], commit list [" + commits + "]";
-        keepOnlyStartingCommitOnInit(commits);
-        updateTranslogDeletionPolicy();
-    }
-
-    /**
-     * Keeping existing unsafe commits when opening an engine can be problematic because these commits are not safe
-     * at the recovering time but they can suddenly become safe in the future.
-     * The following issues can happen if unsafe commits are kept oninit.
-     * <p>
-     * 1. Replica can use unsafe commit in peer-recovery. This happens when a replica with a safe commit c1(max_seqno=1)
-     * and an unsafe commit c2(max_seqno=2) recovers from a primary with c1(max_seqno=1). If a new document(seqno=2)
-     * is added without flushing, the global checkpoint is advanced to 2; and the replica recovers again, it will use
-     * the unsafe commit c2(max_seqno=2 at most gcp=2) as the starting commit for sequenced-based recovery even the
-     * commit c2 contains a stale operation and the document(with seqno=2) will not be replicated to the replica.
-     * <p>
-     * 2. Min translog gen for recovery can go backwards in peer-recovery. This happens when are replica with a safe commit
-     * c1(local_checkpoint=1, recovery_translog_gen=1) and an unsafe commit c2(local_checkpoint=2, recovery_translog_gen=2).
-     * The replica recovers from a primary, and keeps c2 as the last commit, then sets last_translog_gen to 2. Flushing a new
-     * commit on the replica will cause exception as the new last commit c3 will have recovery_translog_gen=1. The recovery
-     * translog generation of a commit is calculated based on the current local checkpoint. The local checkpoint of c3 is 1
-     * while the local checkpoint of c2 is 2.
-     * <p>
-     * 3. Commit without translog can be used in recovery. An old index, which was created before multiple-commits is introduced
-     * (v6.2), may not have a safe commit. If that index has a snapshotted commit without translog and an unsafe commit,
-     * the policy can consider the snapshotted commit as a safe commit for recovery even the commit does not have translog.
-     */
-    private void keepOnlyStartingCommitOnInit(List<? extends IndexCommit> commits) throws IOException {
-        for (IndexCommit commit : commits) {
-            if (startingCommit.equals(commit) == false) {
-                this.deleteCommit(commit);
-            }
+        onCommit(commits);
+        if (safeCommit != commits.get(commits.size() - 1)) {
+            throw new IllegalStateException("Engine is opened, but the last commit isn't safe. Global checkpoint ["
+                + globalCheckpointSupplier.getAsLong() + "], seqNo is last commit ["
+                + SequenceNumbers.loadSeqNoInfoFromLuceneCommit(lastCommit.getUserData().entrySet()) + "], "
+                + "seqNos in safe commit [" + SequenceNumbers.loadSeqNoInfoFromLuceneCommit(safeCommit.getUserData().entrySet()) + "]");
         }
-        assert startingCommit.isDeleted() == false : "Starting commit must not be deleted";
-        lastCommit = startingCommit;
-        safeCommit = startingCommit;
     }
 
     @Override