/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.jackrabbit.oak.plugins.document;

import static com.google.common.collect.Iterables.filter;
import static com.google.common.collect.Lists.newArrayList;
import static com.google.common.collect.Maps.filterKeys;
import static java.util.Collections.singletonList;
import static org.apache.jackrabbit.oak.plugins.document.Collection.JOURNAL;
import static org.apache.jackrabbit.oak.plugins.document.Collection.NODES;
import static org.apache.jackrabbit.oak.plugins.document.util.Utils.PROPERTY_OR_DELETED;
import static org.apache.jackrabbit.oak.plugins.document.util.Utils.isCommitted;
import static org.apache.jackrabbit.oak.plugins.document.util.Utils.resolveCommitRevision;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

import com.google.common.base.Predicate;
import com.google.common.base.Supplier;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;

import org.apache.jackrabbit.oak.commons.TimeDurationFormatter;
import org.apache.jackrabbit.oak.plugins.document.bundlor.DocumentBundlor;
import org.apache.jackrabbit.oak.plugins.document.util.MapFactory;
import org.apache.jackrabbit.oak.plugins.document.util.Utils;
import org.apache.jackrabbit.oak.stats.Clock;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Utility class for recovering potential missing _lastRev updates of nodes due
 * to crash of a node. The recovery agent is also responsible for document
 * sweeping (reverting uncommitted changes).
 * <p>
 * The recovery agent will only sweep documents for a given clusterId if the
 * root document contains a sweep revision for the clusterId. A missing sweep
 * revision for a clusterId indicates an upgrade from an earlier Oak version and
 * a crash before the initial sweep finished. This is not the responsibility of
 * the recovery agent. An initial sweep for an upgrade must either happen with
 * the oak-run 'revisions' sweep command or on startup of an upgraded Oak
 * instance.
 */
public class LastRevRecoveryAgent {
    private final Logger log = LoggerFactory.getLogger(getClass());

    private final DocumentStore store;

    private final RevisionContext revisionContext;

    private final MissingLastRevSeeker missingLastRevUtil;

    private final Consumer<Integer> afterRecovery;

    private static final long LOGINTERVALMS = TimeUnit.MINUTES.toMillis(1);

    public LastRevRecoveryAgent(DocumentStore store,
                                RevisionContext revisionContext,
                                MissingLastRevSeeker seeker,
                                Consumer<Integer> afterRecovery) {
        this.store = store;
        this.revisionContext = revisionContext;
        this.missingLastRevUtil = seeker;
        this.afterRecovery = afterRecovery;
    }

    public LastRevRecoveryAgent(DocumentStore store, RevisionContext context) {
        this(store, context,
                new MissingLastRevSeeker(store, context.getClock()),
                i -> {});
    }

    /**
     * Recover the correct _lastRev updates for potentially missing candidate
     * nodes. If another cluster node is already performing the recovery for the
     * given {@code clusterId}, this method will {@code waitUntil} the given
     * time in milliseconds for the recovery to finish.
     * <p>
     * If recovery is performed for the clusterId as exposed by the revision
     * context passed to the constructor of this recovery agent, then this
     * method will put a deadline on how long recovery may take. The deadline
     * is the current lease end as read from the {@code clusterNodes} collection
     * entry for the {@code clusterId} to recover minus the
     * {@link ClusterNodeInfo#DEFAULT_LEASE_FAILURE_MARGIN_MILLIS}. This method
     * will throw a {@link DocumentStoreException} if the deadline is reached.
     * <p>
     * This method will return:
     * <ul>
     *     <li>{@code -1} when another cluster node is busy performing recovery
     *     for the given {@code clusterId} and the {@code waitUntil} time is
     *     reached.</li>
     *     <li>{@code 0} when no recovery was needed or this thread waited
     *     for another cluster node to finish the recovery within the given
     *     {@code waitUntil} time.</li>
     *     <li>A positive value for the number of recovered documents when
     *     recovery was performed by this thread / cluster node.</li>
     * </ul>
     *
     * @param clusterId the cluster id for which the _lastRev are to be recovered
     * @param waitUntil wait until this time is milliseconds for recovery of the
     *                  given {@code clusterId} if another cluster node is
     *                  already performing the recovery.
     * @return the number of restored nodes or {@code -1} if a timeout occurred
     *          while waiting for an ongoing recovery by another cluster node.
     * @throws DocumentStoreException if the deadline is reached or some other
     *          error occurs while reading from the underlying document store.
     */
    public int recover(int clusterId, long waitUntil)
            throws DocumentStoreException {
        ClusterNodeInfoDocument nodeInfo = missingLastRevUtil.getClusterNodeInfo(clusterId);

        if (nodeInfo != null) {
            // Check if _lastRev recovery needed for this cluster node
            // state is Active && current time past leaseEnd
            if (nodeInfo.isRecoveryNeeded(revisionContext.getClock().getTime())) {
                // retrieve the root document's _lastRev
                NodeDocument root = missingLastRevUtil.getRoot();
                Revision lastRev = root.getLastRev().get(clusterId);
                Revision sweepRev = root.getSweepRevisions().getRevision(clusterId);

                // start time is the _lastRev timestamp of the cluster node
                long startTime;
                String reason;
                //lastRev can be null if other cluster node did not got
                //chance to perform lastRev rollup even once
                if (lastRev != null) {
                    startTime = lastRev.getTimestamp();
                    reason = "lastRev: " + lastRev.toString();
                } else {
                    startTime = nodeInfo.getStartTime();
                    reason = String.format(
                            "no lastRev for root, using startTime %d", startTime);
                }
                if (sweepRev != null && sweepRev.getTimestamp() < startTime) {
                    startTime = sweepRev.getTimestamp();
                    reason = "sweepRev: " + sweepRev.toString();
                }
                // are there branch commits that were merged right before the
                // process crashed and the recovery needs to go further back?
                // go through branch commits before startTime and check if their
                // merge revision is newer than startTime
                Revision bc = getEarliestBranchCommitMergedAround(root, startTime, clusterId);
                if (bc != null) {
                    startTime = bc.getTimestamp();
                    reason = "branchRev: " + bc.toString();
                }

                return recoverCandidates(nodeInfo, startTime, waitUntil, reason);
            }
        }

        log.debug("No recovery needed for clusterId {}", clusterId);
        return 0;
    }

    /**
     * Same as {@link #recover(int, long)}, but does not wait for ongoing
     * recovery.
     *
     * @param clusterId the cluster id for which the _lastRev are to be recovered
     * @return the number of restored nodes or {@code -1} if there is an ongoing
     *          recovery by another cluster node.
     * @throws DocumentStoreException if the deadline is reached or some other
     *          error occurs while reading from the underlying document store.
     */
    public int recover(int clusterId) throws DocumentStoreException {
        return recover(clusterId, 0);
    }

    /**
     * Same as {@link #recover(Iterable, int, boolean)} with {@code dryRun} set
     * to {@code false}.
     *
     * @param suspects the potential suspects
     * @param clusterId the cluster id for which _lastRev recovery needed
     * @return the number of documents that required recovery.
     * @throws DocumentStoreException if the deadline is reached or some other
     *          error occurs while reading from the underlying document store.
     */
    public int recover(Iterable<NodeDocument> suspects, int clusterId)
            throws DocumentStoreException {
        return recover(suspects, clusterId, false);
    }

    /**
     * Recover the correct _lastRev updates for the given candidate nodes. If
     * recovery is performed for the clusterId as exposed by the revision
     * context passed to the constructor of this recovery agent, then this
     * method will put a deadline on how long recovery may take. The deadline
     * is the current lease end as read from the {@code clusterNodes} collection
     * entry for the {@code clusterId} to recover minus the
     * {@link ClusterNodeInfo#DEFAULT_LEASE_FAILURE_MARGIN_MILLIS}. This method
     * will throw a {@link DocumentStoreException} if the deadline is reached.
     *
     * @param suspects the potential suspects
     * @param clusterId the cluster id for which _lastRev recovery needed
     * @param dryRun if {@code true}, this method will only perform a check
     *               but not apply the changes to the _lastRev fields.
     * @return the number of documents that required recovery. This method
     *          returns the number of the affected documents even if
     *          {@code dryRun} is set true and no document was changed.
     * @throws DocumentStoreException if the deadline is reached or some other
     *          error occurs while reading from the underlying document store.
     */
    public int recover(final Iterable<NodeDocument> suspects,
                       final int clusterId, final boolean dryRun)
            throws DocumentStoreException {
        // set a deadline if this is a self recovery. Self recovery does not
        // update the lease in a background thread and must terminate before
        // the lease acquired by the recovery lock expires.
        long deadline = Long.MAX_VALUE;
        if (clusterId == revisionContext.getClusterId()) {
            ClusterNodeInfoDocument nodeInfo = missingLastRevUtil.getClusterNodeInfo(clusterId);
            if (nodeInfo != null && nodeInfo.isActive()) {
                deadline = nodeInfo.getLeaseEndTime() - ClusterNodeInfo.DEFAULT_LEASE_FAILURE_MARGIN_MILLIS;
            }
        }

        NodeDocument rootDoc = Utils.getRootDocument(store);

        // first run a sweep
        final AtomicReference<Revision> sweepRev = new AtomicReference<>();
        if (rootDoc.getSweepRevisions().getRevision(clusterId) != null) {
            // only run a sweep for a cluster node that already has a
            // sweep revision. Initial sweep is not the responsibility
            // of the recovery agent.
            final RevisionContext context = new RecoveryContext(rootDoc,
                    revisionContext.getClock(), clusterId,
                    revisionContext::getCommitValue);
            final NodeDocumentSweeper sweeper = new NodeDocumentSweeper(context, true);
            sweeper.sweep(suspects, new NodeDocumentSweepListener() {
                @Override
                public void sweepUpdate(Map<Path, UpdateOp> updates)
                        throws DocumentStoreException {
                    if (dryRun) {
                        log.info("Dry run of sweeper identified [{}] documents for " +
                                        "cluster node [{}]: {}", updates.size(), clusterId,
                                updates.values());
                        return;
                    }
                    // create an invalidate entry
                    JournalEntry inv = JOURNAL.newDocument(store);
                    inv.modified(updates.keySet());
                    Revision r = context.newRevision().asBranchRevision();
                    UpdateOp invOp = inv.asUpdateOp(r);
                    // and reference it from a regular entry
                    JournalEntry entry = JOURNAL.newDocument(store);
                    entry.invalidate(Collections.singleton(r));
                    Revision jRev = context.newRevision();
                    UpdateOp jOp = entry.asUpdateOp(jRev);
                    if (!store.create(JOURNAL, newArrayList(invOp, jOp))) {
                        String msg = "Unable to create journal entries for " +
                                "document invalidation.";
                        throw new DocumentStoreException(msg);
                    }
                    sweepRev.set(Utils.max(sweepRev.get(), jRev));
                    // now that journal entry is in place, perform the actual
                    // updates on the documents
                    store.createOrUpdate(NODES, newArrayList(updates.values()));
                    log.info("Sweeper updated {}", updates.keySet());
                }
            });

            if (sweepRev.get() != null) {
                // One or more journal entries were created by the sweeper.
                // Make sure the sweep revision is different / newer than the
                // last journal entry written so far. UnsavedModification
                // further down needs a new revision for its journal entry.
                sweepRev.set(Utils.max(sweepRev.get(), context.newRevision()));
            }
        }

        // now deal with missing _lastRev updates
        UnsavedModifications unsaved = new UnsavedModifications();
        UnsavedModifications unsavedParents = new UnsavedModifications();

        //Map of known last rev of checked paths
        Map<Path, Revision> knownLastRevOrModification = MapFactory.getInstance().create();
        final JournalEntry changes = JOURNAL.newDocument(store);

        Clock clock = revisionContext.getClock();

        long totalCount = 0;
        long lastCount = 0;
        long startOfScan = clock.getTime();
        long lastLog = startOfScan;

        for (NodeDocument doc : suspects) {
            totalCount++;
            lastCount++;

            long now = clock.getTime();
            long lastElapsed = now - lastLog;
            if (lastElapsed >= LOGINTERVALMS) {
                TimeDurationFormatter df = TimeDurationFormatter.forLogging();

                long totalElapsed = now - startOfScan;
                long totalRateMin = (totalCount * TimeUnit.MINUTES.toMillis(1)) / totalElapsed;
                long lastRateMin = (lastCount * TimeUnit.MINUTES.toMillis(1)) / lastElapsed;

                String message = String.format(
                        "Recovery for cluster node [%d]: %d nodes scanned in %s (~%d/m) - last interval %d nodes in %s (~%d/m)",
                        clusterId, totalCount, df.format(totalElapsed, TimeUnit.MILLISECONDS), totalRateMin, lastCount,
                        df.format(lastElapsed, TimeUnit.MILLISECONDS), lastRateMin);

                log.info(message);
                lastLog = now;
                lastCount = 0;
            }

            Revision currentLastRev = doc.getLastRev().get(clusterId);

            // 1. determine last committed modification on document
            Revision lastModifiedRev = determineLastModification(doc, clusterId);

            Revision lastRevForParents = Utils.max(lastModifiedRev, currentLastRev);
            // remember the higher of the two revisions. this is the
            // most recent revision currently obtained from either a
            // _lastRev entry or an explicit modification on the document
            if (lastRevForParents != null) {
                knownLastRevOrModification.put(doc.getPath(), lastRevForParents);
            }

            //If both currentLastRev and lostLastRev are null it means
            //that no change is done by suspect cluster on this document
            //so nothing needs to be updated. Probably it was only changed by
            //other cluster nodes. If this node is parent of any child node which
            //has been modified by cluster then that node roll up would
            //add this node path to unsaved

            //2. Update lastRev for parent paths aka rollup
            if (lastRevForParents != null) {
                Path path = doc.getPath();
                changes.modified(path); // track all changes
                while (true) {
                    path = path.getParent();
                    if (path == null) {
                        break;
                    }
                    unsavedParents.put(path, lastRevForParents);
                }
            }
        }

        for (Path parentPath : unsavedParents.getPaths()) {
            Revision calcLastRev = unsavedParents.get(parentPath);
            Revision knownLastRev = knownLastRevOrModification.get(parentPath);
            if (knownLastRev == null) {
                List<Path> missingDocuments = new ArrayList<>();
                // we don't know when the document was last modified with
                // the given clusterId. need to read from store
                NodeDocument doc = findNearestAncestorOrSelf(parentPath, missingDocuments);
                if (doc != null) {
                    Revision lastRev = doc.getLastRev().get(clusterId);
                    Revision lastMod = determineLastModification(doc, clusterId);
                    knownLastRev = Utils.max(lastRev, lastMod);

                    if (!missingDocuments.isEmpty()
                            && doc.getLocalMap(DocumentBundlor.META_PROP_PATTERN).isEmpty()) {
                        // there are missing document and the returned document
                        // does not have bundled nodes
                        for (Path p : missingDocuments) {
                            log.warn("Unable to find document: {}", Utils.getIdFromPath(p));
                        }
                    }
                }
            }

            //Copy the calcLastRev of parent only if they have changed
            //In many case it might happen that parent have consistent lastRev
            //This check ensures that unnecessary updates are not made
            if (knownLastRev == null
                    || calcLastRev.compareRevisionTime(knownLastRev) > 0) {
                unsaved.put(parentPath, calcLastRev);
            }
        }

        if (sweepRev.get() != null) {
            unsaved.put(Path.ROOT, sweepRev.get());
        }

        // take the root's lastRev
        final Revision lastRootRev = unsaved.get(Path.ROOT);

        //Note the size before persist as persist operation
        //would empty the internal state
        int size = unsaved.getPaths().size();
        String updates = unsaved.toString();

        if (dryRun) {
            log.info("Dry run of lastRev recovery identified [{}] documents for " +
                    "cluster node [{}]: {}", size, clusterId, updates);
        } else {
            // check deadline before the update
            if (clock.getTime() > deadline) {
                String msg = String.format("Cluster node %d was unable to " +
                        "perform lastRev recovery for clusterId %d within " +
                        "deadline: %s", clusterId, clusterId,
                        Utils.timestampToString(deadline));
                throw new DocumentStoreException(msg);
            }

            //UnsavedModifications is designed to be used in concurrent
            //access mode. For recovery case there is no concurrent access
            //involve so just pass a new lock instance

            // the lock uses to do the persisting is a plain reentrant lock
            // thus it doesn't matter, where exactly the check is done
            // as to whether the recovered lastRev has already been
            // written to the journal.
            unsaved.persist(store, new Supplier<Revision>() {
                @Override
                public Revision get() {
                    return sweepRev.get();
                }
            }, new UnsavedModifications.Snapshot() {

                @Override
                public void acquiring(Revision mostRecent) {
                    if (lastRootRev == null) {
                        // this should never happen - when unsaved has no changes
                        // that is reflected in the 'map' to be empty - in that
                        // case 'persist()' quits early and never calls
                        // acquiring() here.
                        //
                        // but even if it would occur - if we have no lastRootRev
                        // then we cannot and probably don't have to persist anything
                        return;
                    }

                    final String id = JournalEntry.asId(lastRootRev); // lastRootRev never null at this point
                    final JournalEntry existingEntry = store.find(Collection.JOURNAL, id);
                    if (existingEntry != null) {
                        // then the journal entry was already written - as can happen if
                        // someone else (or the original instance itself) wrote the
                        // journal entry, then died.
                        // in this case, don't write it again.
                        // hence: nothing to be done here. return.
                        log.warn("Journal entry {} already exists", id);
                        return;
                    }

                    // otherwise store a new journal entry now
                    if (store.create(JOURNAL, singletonList(changes.asUpdateOp(lastRootRev)))) {
                        log.info("Recovery created journal entry {}", id);
                    } else {
                        log.warn("Unable to create journal entry {} (already exists).", id);
                    }
                }
            }, new ReentrantLock());

            log.info("Updated lastRev of [{}] documents while performing lastRev recovery for " +
                    "cluster node [{}]: {}", size, clusterId, updates);
        }

        return size;
    }

    //--------------------------< internal >------------------------------------

    /**
     * Get the earliest branch commit before {@code timeMillis} that has been
     * merged after {@code timeMillis}. This method only considers branch
     * commits performed by {@code clusterId}.
     *
     * @param doc the document to check for branch commits.
     * @param timeMillis a time in milliseconds since start of the epoch.
     * @param clusterId a clusterId.
     * @return earliest branch commit or {@code null} if there is none matching
     *          the criteria.
     */
    @Nullable
    private Revision getEarliestBranchCommitMergedAround(@NotNull NodeDocument doc,
                                                         long timeMillis,
                                                         int clusterId) {
        Revision earliest = null;
        for (Revision bc : doc.getLocalBranchCommits()) {
            if (bc.getClusterId() != clusterId) {
                continue;
            }
            String cv = revisionContext.getCommitValue(bc, doc);
            if (isCommitted(cv)) {
                Revision mergeRevision = resolveCommitRevision(bc, cv);
                if (mergeRevision.getTimestamp() > timeMillis
                        && bc.getTimestamp() < timeMillis) {
                    earliest = Utils.min(earliest, bc);
                }
            }
        }
        return earliest;
    }

    @Nullable
    private NodeDocument findNearestAncestorOrSelf(@NotNull Path path,
                                                   @NotNull List<Path> missingDocuments) {
        NodeDocument ancestor;
        for (;;) {
            ancestor = store.find(NODES, Utils.getIdFromPath(path));
            if (ancestor != null) {
                break;
            }
            missingDocuments.add(path);
            path = path.getParent();
            if (path == null) {
                break;
            }
        }
        return ancestor;
    }

    /**
     * Retrieves possible candidates which have been modified after the given
     * {@code startTime} and recovers the missing updates.
     * <p>
     * If recovery is performed for the clusterId as exposed by the revision
     * context passed to the constructor of this recovery agent, then this
     * method will put a deadline on how long recovery may take. The deadline
     * is the current lease end as read from the {@code clusterNodes} collection
     * entry for the {@code clusterId} to recover minus the
     * {@link ClusterNodeInfo#DEFAULT_LEASE_FAILURE_MARGIN_MILLIS}. This method
     * will throw a {@link DocumentStoreException} if the deadline is reached.
     *
     * @param nodeInfo the info of the cluster node to recover.
     * @param startTime the start time
     * @param waitUntil wait at most until this time for an ongoing recovery
     *                  done by another cluster node.
     * @param info a string with additional information how recovery is run.
     * @return the number of restored nodes or {@code -1} if recovery is still
     *      ongoing by another process even when {@code waitUntil} time was
     *      reached.
     * @throws DocumentStoreException if the deadline is reached or some other
     *          error occurs while reading from the underlying document store.
     */
    private int recoverCandidates(final ClusterNodeInfoDocument nodeInfo,
                                  final long startTime,
                                  final long waitUntil,
                                  final String info)
            throws DocumentStoreException {
        ClusterNodeInfoDocument infoDoc = nodeInfo;
        int clusterId = infoDoc.getClusterId();
        for (;;) {
            if (missingLastRevUtil.acquireRecoveryLock(
                    clusterId, revisionContext.getClusterId())) {
                break;
            }

            Clock clock = revisionContext.getClock();
            long remaining = waitUntil - clock.getTime();
            if (remaining < 0) {
                // no need to wait for lock release, waitUntil already reached
                return -1;
            }

            log.info("Last revision recovery already being performed by " +
                    "cluster node {}. Waiting at most until {} for recovery " +
                    "to finish ({} seconds remaining).",
                    infoDoc.getRecoveryBy(), Utils.timestampToString(waitUntil),
                    remaining / 1000);
            // check once every five seconds
            long time = Math.min(waitUntil, clock.getTime() + 5000);
            try {
                clock.waitUntil(time);
            } catch (InterruptedException e) {
                Thread.interrupted();
                String msg = "Interrupted while waiting for _lastRev recovery to finish.";
                throw new DocumentStoreException(msg, e);
            }
            infoDoc = missingLastRevUtil.getClusterNodeInfo(clusterId);
            if (infoDoc == null) {
                String msg = String.format("No cluster node info document " +
                        "for id %d", clusterId);
                throw new DocumentStoreException(msg);
            }
            if (!infoDoc.isRecoveryNeeded(clock.getTime())) {
                // meanwhile another process finished recovery
                return 0;
            }
        }

        // if we get here, the recovery lock was acquired successfully
        boolean success = false;
        try {
            log.info("Recovering candidates modified after: [{}] for clusterId [{}] [{}]",
                    Utils.timestampToString(startTime), clusterId, info);

            Iterable<NodeDocument> suspects = missingLastRevUtil.getCandidates(startTime);
            try {
                log.info("Performing Last Revision Recovery for clusterNodeId {}", clusterId);
                int num = recover(suspects, clusterId);
                success = true;
                return num;
            } finally {
                Utils.closeIfCloseable(suspects);
            }
        } finally {
            missingLastRevUtil.releaseRecoveryLock(clusterId, success);
            afterRecovery.accept(clusterId);
        }
    }

    /**
     * Determines the last committed modification to the given document by
     * a {@code clusterId}.
     *
     * @param doc a document.
     * @param clusterId clusterId for which the last committed modification is
     *                  looked up.
     * @return the commit revision of the last modification by {@code clusterId}
     *          to the given document.
     */
    @Nullable
    private Revision determineLastModification(NodeDocument doc, int clusterId) {
        ClusterPredicate cp = new ClusterPredicate(clusterId);

        Revision lastModified = null;
        for (String property : Sets.filter(doc.keySet(), PROPERTY_OR_DELETED)) {
            Map<Revision, String> valueMap = doc.getLocalMap(property);
            // collect committed changes of this cluster node
            for (Map.Entry<Revision, String> entry : filterKeys(valueMap, cp).entrySet()) {
                Revision rev = entry.getKey();
                String cv = revisionContext.getCommitValue(rev, doc);
                if (isCommitted(cv)) {
                    lastModified = Utils.max(lastModified, resolveCommitRevision(rev, cv));
                    break;
                }
            }
        }
        return lastModified;
    }

    /**
     * Determines if any of the cluster node failed to renew its lease and
     * did not properly shutdown. If any such cluster node is found then are
     * potential candidates for last rev recovery. This method also returns
     * true when there is a cluster node with an ongoing recovery.
     *
     * @return true if last rev recovery needs to be performed for any of the
     *          cluster nodes
     */
    public boolean isRecoveryNeeded(){
        return missingLastRevUtil.isRecoveryNeeded();
    }

    public void performRecoveryIfNeeded() {
        if (isRecoveryNeeded()) {
            Iterable<Integer> clusterIds = getRecoveryCandidateNodes();
            log.info("ClusterNodeId [{}] starting Last Revision Recovery for clusterNodeId(s) {}",
                    revisionContext.getClusterId(), clusterIds);
            for (int clusterId : clusterIds) {
                if (recover(clusterId) == -1) {
                    log.info("Last Revision Recovery for cluster node {} " +
                            "ongoing by other cluster node.", clusterId);
                }
            }
        }
    }

    /**
     * Gets the _lastRev recovery candidate cluster nodes. This also includes
     * cluster nodes that are currently being recovered. The method would not
     * return self as a candidate for recovery even if it has failed to update
     * lease in time
     *
     * @return the recovery candidate nodes.
     */
    public Iterable<Integer> getRecoveryCandidateNodes() {
        return Iterables.transform(filter(missingLastRevUtil.getAllClusters(),
                new Predicate<ClusterNodeInfoDocument>() {
            @Override
            public boolean apply(ClusterNodeInfoDocument input) {
                return revisionContext.getClusterId() != input.getClusterId()
                        && input.isRecoveryNeeded(revisionContext.getClock().getTime());
            }
        }), ClusterNodeInfoDocument::getClusterId);
    }

    private static class ClusterPredicate implements Predicate<Revision> {
        private final int clusterId;

        private ClusterPredicate(int clusterId) {
            this.clusterId = clusterId;
        }

        @Override
        public boolean apply(Revision input) {
            return clusterId == input.getClusterId();
        }
    }
}