Retain history for peer recovery using leases

server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java

@@ -104,6 +104,7 @@ public void execute() throws Exception {
         pendingActions.incrementAndGet(); // increase by 1 until we finish all primary coordination
         primaryResult = primary.perform(request);
         primary.updateLocalCheckpointForShard(primaryRouting.allocationId().getId(), primary.localCheckpoint());
+        primary.updateLocalCheckpointOfSafeCommitForShard(primaryRouting.allocationId().getId(), primary.localCheckpointOfSafeCommit());
         final ReplicaRequest replicaRequest = primaryResult.replicaRequest();
         if (replicaRequest != null) {
             if (logger.isTraceEnabled()) {
@@ -170,6 +171,7 @@ public void onResponse(ReplicaResponse response) {
                 try {
                     primary.updateLocalCheckpointForShard(shard.allocationId().getId(), response.localCheckpoint());
                     primary.updateGlobalCheckpointForShard(shard.allocationId().getId(), response.globalCheckpoint());
+                    primary.updateLocalCheckpointOfSafeCommitForShard(shard.allocationId().getId(), response.localCheckpointOfSafeCommit());
                 } catch (final AlreadyClosedException e) {
                     // okay, the index was deleted or this shard was never activated after a relocation; fall through and finish normally
                 } catch (final Exception e) {
@@ -334,6 +336,14 @@ public interface Primary<
          */
         void updateGlobalCheckpointForShard(String allocationId, long globalCheckpoint);
 
+        /**
+         * Update the local knowledge of the local checkpoint of the safe commit for the specified allocation ID.
+         *
+         * @param allocationId     the allocation ID to update the local checkpoint of the safe commit for
+         * @param localCheckpointOfSafeCommit the local checkpoint of the safe commit
+         */
+        void updateLocalCheckpointOfSafeCommitForShard(String allocationId, long localCheckpointOfSafeCommit);
+
         /**
          * Returns the local checkpoint on the primary shard.
          *
@@ -348,6 +358,13 @@ public interface Primary<
          */
         long globalCheckpoint();
 
+        /**
+         * Returns the local checkpoint of the safe commit on the primary shard.
+         *
+         * @return the local checkpoint of the safe commit
+         */
+        long localCheckpointOfSafeCommit();
+
         /**
          * Returns the maximum seq_no of updates (index operations overwrite Lucene) or deletes on the primary.
          * This value must be captured after the execution of a replication request on the primary is completed.
@@ -423,6 +440,12 @@ public interface ReplicaResponse {
          **/
         long globalCheckpoint();
 
+        /**
+         * The local checkpoint of the safe commit for the shard copy.
+         *
+         * @return the local checkpoint of the safe commit
+         **/
+        long localCheckpointOfSafeCommit();
     }
 
     public static class RetryOnPrimaryException extends ElasticsearchException {

server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java

@@ -23,6 +23,7 @@
 import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.ActionResponse;
@@ -231,6 +232,13 @@ protected boolean resolveIndex() {
         return true;
     }
 
+    /**
+     * True if this action can be replicated even on closed indices.
+     */
+    protected boolean isReplicatedOnClosedIndices() {
+        return false;
+    }
+
     protected TransportRequestOptions transportOptions(Settings settings) {
         return TransportRequestOptions.EMPTY;
     }
@@ -618,7 +626,8 @@ public void onResponse(Releasable releasable) {
                 final ReplicaResult replicaResult = shardOperationOnReplica(request, replica);
                 releasable.close(); // release shard operation lock before responding to caller
                 final TransportReplicationAction.ReplicaResponse response =
-                        new ReplicaResponse(replica.getLocalCheckpoint(), replica.getGlobalCheckpoint());
+                        new ReplicaResponse(replica.getLocalCheckpoint(), replica.getGlobalCheckpoint(),
+                                            replica.getLocalCheckpointOfSafeCommit());
                 replicaResult.respond(new ResponseListener(response));
             } catch (final Exception e) {
                 Releasables.closeWhileHandlingException(releasable); // release shard operation lock before responding to caller
@@ -773,7 +782,7 @@ protected void doRun() {
                     retry(new IndexNotFoundException(concreteIndex));
                     return;
                 }
-                if (indexMetaData.getState() == IndexMetaData.State.CLOSE) {
+                if (isReplicatedOnClosedIndices() == false && indexMetaData.getState() == IndexMetaData.State.CLOSE) {
                     throw new IndexClosedException(indexMetaData.getIndex());
                 }
 
@@ -1045,6 +1054,11 @@ public void updateGlobalCheckpointForShard(final String allocationId, final long
             indexShard.updateGlobalCheckpointForShard(allocationId, globalCheckpoint);
         }
 
+        @Override
+        public void updateLocalCheckpointOfSafeCommitForShard(final String allocationId, final long localCheckpointOfSafeCommit) {
+            indexShard.updateLocalCheckpointOfSafeCommitForShard(allocationId, localCheckpointOfSafeCommit);
+        }
+
         @Override
         public long localCheckpoint() {
             return indexShard.getLocalCheckpoint();
@@ -1055,6 +1069,11 @@ public long globalCheckpoint() {
             return indexShard.getGlobalCheckpoint();
         }
 
+        @Override
+        public long localCheckpointOfSafeCommit() {
+            return indexShard.getLocalCheckpointOfSafeCommit();
+        }
+
         @Override
         public long maxSeqNoOfUpdatesOrDeletes() {
             return indexShard.getMaxSeqNoOfUpdatesOrDeletes();
@@ -1070,12 +1089,13 @@ public ReplicationGroup getReplicationGroup() {
     public static class ReplicaResponse extends ActionResponse implements ReplicationOperation.ReplicaResponse {
         private long localCheckpoint;
         private long globalCheckpoint;
+        private long localCheckpointOfSafeCommit;
 
         ReplicaResponse() {
 
         }
 
-        public ReplicaResponse(long localCheckpoint, long globalCheckpoint) {
+        public ReplicaResponse(long localCheckpoint, long globalCheckpoint, long localCheckpointOfSafeCommit) {
             /*
              * A replica should always know its own local checkpoints so this should always be a valid sequence number or the pre-6.0
              * checkpoint value when simulating responses to replication actions that pre-6.0 nodes are not aware of (e.g., the global
@@ -1084,20 +1104,29 @@ public ReplicaResponse(long localCheckpoint, long globalCheckpoint) {
             assert localCheckpoint != SequenceNumbers.UNASSIGNED_SEQ_NO;
             this.localCheckpoint = localCheckpoint;
             this.globalCheckpoint = globalCheckpoint;
+            this.localCheckpointOfSafeCommit = localCheckpointOfSafeCommit;
         }
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
             super.readFrom(in);
             localCheckpoint = in.readZLong();
             globalCheckpoint = in.readZLong();
+            if (in.getVersion().onOrAfter(Version.V_8_0_0)) { // TODO V_7_0_0 for backport
+                localCheckpointOfSafeCommit = in.readZLong();
+            } else {
+                localCheckpointOfSafeCommit = SequenceNumbers.UNASSIGNED_SEQ_NO;
+            }
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeZLong(localCheckpoint);
             out.writeZLong(globalCheckpoint);
+            if (out.getVersion().onOrAfter(Version.V_8_0_0)) { // TODO V_7_0_0 for backport
+                out.writeZLong(localCheckpointOfSafeCommit);
+            }
         }
 
         @Override
@@ -1110,18 +1139,24 @@ public long globalCheckpoint() {
             return globalCheckpoint;
         }
 
+        @Override
+        public long localCheckpointOfSafeCommit() {
+            return localCheckpointOfSafeCommit;
+        }
+
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             ReplicaResponse that = (ReplicaResponse) o;
             return localCheckpoint == that.localCheckpoint &&
-                globalCheckpoint == that.globalCheckpoint;
+                globalCheckpoint == that.globalCheckpoint &&
+                localCheckpointOfSafeCommit == that.localCheckpointOfSafeCommit;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(localCheckpoint, globalCheckpoint);
+            return Objects.hash(localCheckpoint, globalCheckpoint, localCheckpointOfSafeCommit);
         }
     }
 

server/src/main/java/org/elasticsearch/index/IndexService.java

@@ -26,6 +26,7 @@
 import org.apache.lucene.util.Accountable;
 import org.elasticsearch.Assertions;
 import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -88,6 +89,7 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
@@ -122,6 +124,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private volatile AsyncTranslogFSync fsyncTask;
     private volatile AsyncGlobalCheckpointTask globalCheckpointTask;
     private volatile AsyncRetentionLeaseSyncTask retentionLeaseSyncTask;
+    private volatile AsyncPeerRecoveryRetentionLeaseRenewalTask peerRecoveryRetentionLeaseRenewalTask;
 
     // don't convert to Setting<> and register... we only set this in tests and register via a plugin
     private final String INDEX_TRANSLOG_RETENTION_CHECK_INTERVAL_SETTING = "index.translog.retention.check_interval";
@@ -199,6 +202,7 @@ public IndexService(
         this.trimTranslogTask = new AsyncTrimTranslogTask(this);
         this.globalCheckpointTask = new AsyncGlobalCheckpointTask(this);
         this.retentionLeaseSyncTask = new AsyncRetentionLeaseSyncTask(this);
+        this.peerRecoveryRetentionLeaseRenewalTask = new AsyncPeerRecoveryRetentionLeaseRenewalTask(this);
         rescheduleFsyncTask(indexSettings.getTranslogDurability());
     }
 
@@ -289,7 +293,8 @@ public synchronized void close(final String reason, boolean delete) throws IOExc
                         fsyncTask,
                         trimTranslogTask,
                         globalCheckpointTask,
-                        retentionLeaseSyncTask);
+                        retentionLeaseSyncTask,
+                        peerRecoveryRetentionLeaseRenewalTask);
             }
         }
     }
@@ -317,8 +322,10 @@ private long getAvgShardSizeInBytes() throws IOException {
     public synchronized IndexShard createShard(
             final ShardRouting routing,
             final Consumer<ShardId> globalCheckpointSyncer,
-            final RetentionLeaseSyncer retentionLeaseSyncer) throws IOException {
+            final RetentionLeaseSyncer retentionLeaseSyncer,
+            final BiConsumer<ShardId, ActionListener<Void>> peerRecoveryRetentionLeaseRenewer) throws IOException {
         Objects.requireNonNull(retentionLeaseSyncer);
+        Objects.requireNonNull(peerRecoveryRetentionLeaseRenewer);
         /*
          * TODO: we execute this in parallel but it's a synced method. Yet, we might
          * be able to serialize the execution via the cluster state in the future. for now we just
@@ -408,7 +415,8 @@ public synchronized IndexShard createShard(
                     indexingOperationListeners,
                     () -> globalCheckpointSyncer.accept(shardId),
                     retentionLeaseSyncer,
-                    circuitBreakerService);
+                    circuitBreakerService,
+                    listener -> peerRecoveryRetentionLeaseRenewer.accept(shardId, listener));
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
             eventListener.afterIndexShardCreated(indexShard);
             shards = newMapBuilder(shards).put(shardId.id(), indexShard).immutableMap();
@@ -795,6 +803,10 @@ private void syncRetentionLeases() {
         }
     }
 
+    private void renewPeerRecoveryRetentionLeases() {
+        sync(IndexShard::renewPeerRecoveryRetentionLeases, "peer recovery retention leases");
+    }
+
     private void sync(final Consumer<IndexShard> sync, final String source) {
         for (final IndexShard shard : this.shards.values()) {
             if (shard.routingEntry().active() && shard.routingEntry().primary()) {
@@ -937,6 +949,15 @@ public String toString() {
                     Property.Dynamic,
                     Property.IndexScope);
 
+    // this setting is intentionally not registered, it is only used in tests
+    public static final Setting<TimeValue> RETENTION_LEASE_PEER_RECOVERY_SYNC_INTERVAL_SETTING =
+        Setting.timeSetting(
+            "index.soft_deletes.retention_lease.peer_recovery.sync_interval",
+            new TimeValue(5, TimeUnit.MINUTES),
+            new TimeValue(0, TimeUnit.MILLISECONDS),
+            Property.Dynamic,
+            Property.IndexScope);
+
     /**
      * Background task that syncs the global checkpoint to replicas.
      */
@@ -986,6 +1007,30 @@ public String toString() {
 
     }
 
+    final class AsyncPeerRecoveryRetentionLeaseRenewalTask extends BaseAsyncTask {
+
+        // TODO perhaps we can just piggyback on the synced flush that happens after 5 minutes of inactivity instead?
+
+        AsyncPeerRecoveryRetentionLeaseRenewalTask(final IndexService indexService) {
+            super(indexService, RETENTION_LEASE_PEER_RECOVERY_SYNC_INTERVAL_SETTING.get(indexService.getIndexSettings().getSettings()));
+        }
+
+        @Override
+        protected void runInternal() {
+            indexService.renewPeerRecoveryRetentionLeases();
+        }
+
+        @Override
+        protected String getThreadPool() {
+            return ThreadPool.Names.MANAGEMENT;
+        }
+
+        @Override
+        public String toString() {
+            return "peer_recovery_retention_lease_sync";
+        }
+    }
+
     AsyncRefreshTask getRefreshTask() { // for tests
         return refreshTask;
     }

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

@@ -264,6 +264,8 @@ protected final DocsStats docsStats(IndexReader indexReader) {
         return new DocsStats(numDocs, numDeletedDocs, sizeInBytes);
     }
 
+    public abstract long getLocalCheckpointOfSafeCommit();
+
     /**
      * Performs the pre-closing checks on the {@link Engine}.
      *
@@ -766,7 +768,7 @@ public abstract int estimateNumberOfHistoryOperations(String source,
                                                                 MapperService mapperService, long startingSeqNo) throws IOException;
 
     /**
-     * Checks if this engine has every operations since  {@code startingSeqNo}(inclusive) in its translog
+     * Checks if this engine has every operations since  {@code startingSeqNo}(inclusive) in its history (either Lucene or translog)
      */
     public abstract boolean hasCompleteOperationHistory(String source, MapperService mapperService, long startingSeqNo) throws IOException;