/*
 * 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.base.Preconditions.checkNotNull;
import static org.apache.jackrabbit.oak.plugins.document.ClusterNodeInfo.ClusterNodeState.ACTIVE;
import static org.apache.jackrabbit.oak.plugins.document.ClusterNodeInfo.ClusterNodeState.NONE;
import static org.apache.jackrabbit.oak.plugins.document.ClusterNodeInfo.RecoverLockState.ACQUIRED;
import static org.apache.jackrabbit.oak.plugins.document.util.Utils.getModuleVersion;

import java.lang.management.ManagementFactory;
import java.net.NetworkInterface;
import java.time.Instant;
import java.time.format.DateTimeFormatter;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import com.google.common.base.Stopwatch;

import org.apache.jackrabbit.oak.commons.StringUtils;
import org.apache.jackrabbit.oak.plugins.document.util.SystemPropertySupplier;
import org.apache.jackrabbit.oak.stats.Clock;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;

/**
 * Information about a cluster node.
 */
public class ClusterNodeInfo {

    private static final String LEASE_CHECK_FAILED_MSG = "This oak instance failed to update "
            + "the lease in time and can therefore no longer access this DocumentNodeStore.";

    private static final Logger LOG = LoggerFactory.getLogger(ClusterNodeInfo.class);

    /**
     * The prefix for random (non-reusable) keys.
     */
    private static final String RANDOM_PREFIX = "random:";

    /**
     * The machine id.
     */
    private static final String MACHINE_ID_KEY = "machine";

    /**
     * The Oak version.
     */
    static final String OAK_VERSION_KEY = "oakVersion";

    /**
     * The unique instance id within this machine (the current working directory
     * if not set).
     */
    private static final String INSTANCE_ID_KEY = "instance";

    /**
     * The end of the lease.
     */
    public static final String LEASE_END_KEY = "leaseEnd";

    /**
     * The start time.
     */
    public static final String START_TIME_KEY = "startTime";

    /**
     * The key for the root-revision of the last background write (of unsaved
     * modifications) - that is: the last root-revision written by the instance
     * in case of a clear shutdown or via recovery of another instance in case
     * of a crash
     */
    public static final String LAST_WRITTEN_ROOT_REV_KEY = "lastWrittenRootRev";

    /**
     * The state of the cluster. On proper shutdown the state should be cleared.
     *
     * @see org.apache.jackrabbit.oak.plugins.document.ClusterNodeInfo.ClusterNodeState
     */
    public static final String STATE = "state";
    
    /**
     * The broadcast id. If the broadcasting cache is used, a new id is set after startup.
     */
    public static final String BROADCAST_ID = "broadcastId";

    /**
     * The broadcast listener (host:port). If the broadcasting cache is used, this is set after startup.
     */
    public static final String BROADCAST_LISTENER = "broadcastListener";

    public static enum ClusterNodeState {
        NONE,
        /**
         * Indicates that cluster node is currently active
         */
        ACTIVE;

        static ClusterNodeState fromString(String state) {
            if (state == null) {
                return NONE;
            }
            return valueOf(state);
        }
    }

    /**
     * Flag to indicate whether the _lastRev recovery is in progress.
     *
     * @see RecoverLockState
     */
    public static final String REV_RECOVERY_LOCK = "recoveryLock";

    public static enum RecoverLockState {
        NONE,
        /**
         * _lastRev recovery in progress
         */
        ACQUIRED;

        static RecoverLockState fromString(String state) {
            if (state == null) {
                return NONE;
            }
            return valueOf(state);
        }
    }

    /**
     * Flag indicating which cluster node is running the recovery.
     */
    public static final String REV_RECOVERY_BY = "recoveryBy";

    /**
     * Additional info, such as the process id, for support.
     */
    private static final String INFO_KEY = "info";

    /**
     * The read/write mode key. Specifies the read/write preference to be used with
     * DocumentStore
     */
    private static final String READ_WRITE_MODE_KEY = "readWriteMode";

    /**
     * Key for invisible flag
     */
    public static final String INVISIBLE = "invisible";

    /**
     * The unique machine id (the MAC address if available).
     */
    private static final String MACHINE_ID = getHardwareMachineId();

    /**
     * The process id (if available).
     */
    private static final long PROCESS_ID = getProcessId();

    /**
     * The current working directory.
     * Note: marked protected non-final for testing purpose only.
     */
    protected static String WORKING_DIR = System.getProperty("user.dir", "");

    /**
     * <b>Only Used For Testing</b>
     */
    private static Clock clock = Clock.SIMPLE;

    /** OAK-3398 : default lease duration 120sec **/
    private static final int DEFAULT_LEASE_DURATION_SEC = SystemPropertySupplier.create("oak.documentMK.leaseDurationSeconds", 120)
            .loggingTo(LOG).validateWith(value -> value >= 0)
            .formatSetMessage((name, value) -> String.format("Lease duration set to: %ss (using system property %s)", name, value)).get();
    public static final int DEFAULT_LEASE_DURATION_MILLIS = 1000 * DEFAULT_LEASE_DURATION_SEC;

    /** OAK-3398 : default update interval 10sec **/
    public static final int DEFAULT_LEASE_UPDATE_INTERVAL_MILLIS = 1000 * 10;

    /** OAK-3398 : default failure margin 20sec before actual lease timeout
     * (note that OAK-3399 / MAX_RETRY_SLEEPS_BEFORE_LEASE_FAILURE eats
     * off another few seconds from this margin, by default 5sec,
     * so the actual default failure-margin is down to 15sec - and that is high-noon!)
     */
    public static final int DEFAULT_LEASE_FAILURE_MARGIN_MILLIS = 1000 * 20;

    public static final boolean DEFAULT_LEASE_CHECK_DISABLED = SystemPropertySupplier
            .create("oak.documentMK.disableLeaseCheck", Boolean.FALSE).loggingTo(LOG).get();

    /**
     * Default lease check mode is strict, unless disabled via system property.
     */
    static final LeaseCheckMode DEFAULT_LEASE_CHECK_MODE =
            DEFAULT_LEASE_CHECK_DISABLED ? LeaseCheckMode.DISABLED : LeaseCheckMode.STRICT;

    /** OAK-3399 : max number of times we're doing a 1sec retry loop just before declaring lease failure **/
    private static final int MAX_RETRY_SLEEPS_BEFORE_LEASE_FAILURE = 5;

    /**
     * The Oak version.
     */
    private static final String OAK_VERSION = getModuleVersion();

    /**
     * The number of milliseconds for a lease (2 minute by default, and
     * initially).
     */
    private long leaseTime = DEFAULT_LEASE_DURATION_MILLIS;

    /**
     * The number of milliseconds after which a lease will be updated
     * (should not be every second as that would increase number of
     * writes towards DocumentStore considerably - but it should also
     * not be too low as that would eat into the lease duration on average.
     */
    private long leaseUpdateInterval = DEFAULT_LEASE_UPDATE_INTERVAL_MILLIS;

    /**
     * The number of milliseconds that a lease must still be valid
     * before prematurely declaring it as failed. The default is 20sec.
     * The idea of declaring a lease as failed before it actually failed
     * is to avoid a race condition where the local instance assumes
     * things are all fine but another instance in the cluster will
     * 'in the same moment' declare it as failed. The lease should be
     * checked every second and updated after 10sec, so it should always
     * have a validity of at least 110sec - if that's down to this margin
     * of 20sec then things are not good and we have to give up.
     */
    private long leaseFailureMargin = DEFAULT_LEASE_FAILURE_MARGIN_MILLIS;

    /**
     * The assigned cluster id.
     */
    private final int id;

    /**
     * The machine id.
     */
    private final String machineId;

    /**
     * The instance id.
     */
    private final String instanceId;

    /**
     * The document store that is used to renew the lease.
     */
    private final DocumentStore store;

    /**
     * The time (in milliseconds UTC) where this instance was started.
     */
    private final long startTime;

    /**
     * A unique id.
     */
    private final String uuid = UUID.randomUUID().toString();

    /**
     * The time (in milliseconds UTC) where the lease of this instance ends.
     */
    private volatile long leaseEndTime;

    /**
     * The value of leaseEnd last updated towards DocumentStore -
     * this one is used to compare against (for OAK-3398) when checking
     * if any other instance updated the lease or if the lease is unchanged.
     * (This is kind of a duplication of the leaseEndTime field, yes - but the semantics
     * are that previousLeaseEndTime exactly only serves the purpose of
     * keeping the value of what was stored in the previous lease update.
     * leaseEndTime on the other hand serves the purpose of *defining the lease end*,
     * these are two different concerns, thus justify two different fields.
     * the leaseEndTime for example can be manipulated during tests therefore,
     * without interfering with previousLeaseEndTime)
     */
    private long previousLeaseEndTime;

    /**
     * The read/write mode.
     */
    private String readWriteMode;

    /**
     * The state of the cluster node.
     */
    private ClusterNodeState state = ACTIVE;

    /**
     * OAK-2739 / OAK-3397 : once a lease check turns out negative, this flag
     * is set to prevent any further checks to succeed. Also, only the first
     * one to change this flag will take the appropriate action that results
     * from a failed leaseCheck (which is currently to stop oak-store-document bundle)
     */
    private boolean leaseCheckFailed = false;

    /**
     * Default lease check mode is strict, unless disabled by
     */
    private LeaseCheckMode leaseCheckMode = DEFAULT_LEASE_CHECK_MODE;

    /**
     * In memory flag indicating that this ClusterNode is entry is new and is being added to
     * DocumentStore for the first time
     *
     * If false then it indicates that a previous entry for current node existed and that is being
     * reused
     */
    private boolean newEntry;

    /**
     * OAK-3397 / OAK-3400 : the LeaseFailureHandler is the one that actually
     * stops the oak-store-document bundle (or does something else if necessary)
     */
    private LeaseFailureHandler leaseFailureHandler;

    /**
     * Flag to indicate this node is invisible to cluster view and thus recovery.
     */
    private boolean invisible;


    private ClusterNodeInfo(int id, DocumentStore store, String machineId,
                            String instanceId, boolean newEntry, boolean invisible) {
        this.id = id;
        this.startTime = getCurrentTime();
        this.leaseEndTime = this.startTime +leaseTime;
        this.previousLeaseEndTime = this.leaseEndTime;
        this.store = store;
        this.machineId = machineId;
        this.instanceId = instanceId;
        this.newEntry = newEntry;
        this.invisible = invisible;
    }

    void setLeaseCheckMode(@NotNull LeaseCheckMode mode) {
        this.leaseCheckMode = checkNotNull(mode);
    }

    LeaseCheckMode getLeaseCheckMode() {
        return leaseCheckMode;
    }
    
    public int getId() {
        return id;
    }

    String getMachineId() {
        return machineId;
    }

    String getInstanceId() {
        return instanceId;
    }

    boolean isInvisible() {
        return invisible;
    }

    /**
     * Create a cluster node info instance to be utilized for read only access
     * to underlying store.
     *
     * @param store the document store.
     * @return the cluster node info
     */
    public static ClusterNodeInfo getReadOnlyInstance(DocumentStore store) {
        return new ClusterNodeInfo(0, store, MACHINE_ID, WORKING_DIR, true, true) {
            @Override
            public void dispose() {
            }

            @Override
            public long getLeaseEndTime() {
                return Long.MAX_VALUE;
            }

            @Override
            public void performLeaseCheck() {
            }

            @Override
            public boolean renewLease() {
                return false;
            }

            @Override
            public void setInfo(Map<String, String> info) {}

            @Override
            public void setLeaseFailureHandler(LeaseFailureHandler leaseFailureHandler) {}
        };
    }

    /**
     * Get or create a cluster node info instance for the store.
     *
     * @param store the document store (for the lease)
     * @param recoveryHandler the recovery handler to call for a clusterId with
     *                        an expired lease.
     * @param machineId the machine id (null for MAC address)
     * @param instanceId the instance id (null for current working directory)
     * @param configuredClusterId the configured cluster id (or 0 for dynamic assignment)
     * @return the cluster node info
     */
    public static ClusterNodeInfo getInstance(DocumentStore store,
        RecoveryHandler recoveryHandler,
        String machineId,
        String instanceId,
        int configuredClusterId) {

        return getInstance(store, recoveryHandler, machineId, instanceId, configuredClusterId, false);
    }

    /**
     * Get or create a cluster node info instance for the store.
     *
     * @param store the document store (for the lease)
     * @param recoveryHandler the recovery handler to call for a clusterId with
     *                        an expired lease.
     * @param machineId the machine id (null for MAC address)
     * @param instanceId the instance id (null for current working directory)
     * @param configuredClusterId the configured cluster id (or 0 for dynamic assignment)
     * @return the cluster node info
     */
    public static ClusterNodeInfo getInstance(DocumentStore store,
                                              RecoveryHandler recoveryHandler,
                                              String machineId,
                                              String instanceId,
                                              int configuredClusterId,
                                              boolean invisible) {
        // defaults for machineId and instanceID
        if (machineId == null) {
            machineId = MACHINE_ID;
        }
        if (instanceId == null) {
            instanceId = WORKING_DIR;
        }

        int retries = 10;
        for (int i = 0; i < retries; i++) {
            Map.Entry<ClusterNodeInfo, Long> suggestedClusterNode =
                    createInstance(store, recoveryHandler, machineId,
                            instanceId, configuredClusterId, i == 0, invisible);
            ClusterNodeInfo clusterNode = suggestedClusterNode.getKey();
            Long currentStartTime = suggestedClusterNode.getValue();
            String key = String.valueOf(clusterNode.id);
            UpdateOp update = new UpdateOp(key, clusterNode.newEntry);
            update.set(MACHINE_ID_KEY, clusterNode.machineId);
            update.set(INSTANCE_ID_KEY, clusterNode.instanceId);
            update.set(LEASE_END_KEY, clusterNode.leaseEndTime);
            update.set(START_TIME_KEY, clusterNode.startTime);
            update.set(INFO_KEY, clusterNode.toString());
            update.set(STATE, ACTIVE.name());
            update.set(OAK_VERSION_KEY, OAK_VERSION);
            update.set(INVISIBLE, invisible);

            ClusterNodeInfoDocument before = null;
            final boolean success;
            if (clusterNode.newEntry) {
                // For new entry do a create. This ensures that if two nodes
                // create entry with same id then only one would succeed
                success = store.create(Collection.CLUSTER_NODES, Collections.singletonList(update));
            } else {
                // remember how the entry looked before the update
                before = store.find(Collection.CLUSTER_NODES, key);

                // perform a conditional update with a check on the startTime
                // field. If there are competing cluster nodes trying to acquire
                // the same inactive clusterId, only one of them will succeed.
                update.equals(START_TIME_KEY, currentStartTime);
                // ensure some other conditions
                // 1) must not be active
                update.notEquals(STATE, ACTIVE.name());
                // 2) must not have a recovery lock
                update.notEquals(REV_RECOVERY_LOCK, ACQUIRED.name());

                success = store.findAndUpdate(Collection.CLUSTER_NODES, update) != null;
            }

            if (success) {
                logClusterIdAcquired(clusterNode, before);
                return clusterNode;
            }
            LOG.info("Collision while acquiring clusterId {}. Retrying...",
                    clusterNode.getId());
        }
        throw new DocumentStoreException("Could not get cluster node info (tried " + retries + " times)");
    }

    private static Map.Entry<ClusterNodeInfo, Long> createInstance(DocumentStore store,
                                                                   RecoveryHandler recoveryHandler,
                                                                   String machineId,
                                                                   String instanceId,
                                                                   int configuredClusterId,
                                                                   boolean waitForLease,
                                                                   boolean invisible) {

        long now = getCurrentTime();
        int maxId = 0;

        ClusterNodeInfoDocument alreadyExistingConfigured = null;
        String reuseFailureReason = "";
        List<ClusterNodeInfoDocument> list = ClusterNodeInfoDocument.all(store);
        Map<Integer, Long> startTimes = new HashMap<>();
        SortedSet<ClusterNodeInfo> candidates = new TreeSet<>(
                new ClusterNodeInfoComparator(machineId, instanceId));

        for (ClusterNodeInfoDocument doc : list) {

            String key = doc.getId();

            int id;
            try {
                id = doc.getClusterId();
            } catch (Exception e) {
                LOG.debug("Skipping cluster node info document {} because ID is invalid", key);
                continue;
            }

            maxId = Math.max(maxId, id);

            // cannot use an entry without start time
            if (doc.getStartTime() == -1) {
                reuseFailureReason = reject(id,
                        "Cluster node entry does not have a startTime. ");
                continue;
            }

            // if a cluster id was configured: check that and abort if it does
            // not match
            if (configuredClusterId != 0) {
                if (configuredClusterId != id) {
                    continue;
                } else {
                    alreadyExistingConfigured = doc;
                }
            }

            Long leaseEnd = (Long) doc.get(LEASE_END_KEY);
            String mId = "" + doc.get(MACHINE_ID_KEY);
            String iId = "" + doc.get(INSTANCE_ID_KEY);

            // handle active clusterId with valid lease and no recovery lock
            // -> potentially wait for lease if machine and instance id match
            if (leaseEnd != null
                    && leaseEnd > now
                    && !doc.isRecoveryNeeded(now)) {
                // wait if (a) instructed to, and (b) also the remaining time
                // time is not much bigger than the lease interval (in which
                // case something is very very wrong anyway)
                if (waitForLease && (leaseEnd - now) < (DEFAULT_LEASE_DURATION_MILLIS + 5000) && mId.equals(machineId)
                        && iId.equals(instanceId)) {
                    boolean worthRetrying = waitForLeaseExpiry(store, doc, leaseEnd, machineId, instanceId);
                    if (worthRetrying) {
                        return createInstance(store, recoveryHandler, machineId, instanceId, configuredClusterId, false, invisible);
                    }
                }

                reuseFailureReason = reject(id,
                        "leaseEnd " + leaseEnd + " > " + now + " - " + (leaseEnd - now) + "ms in the future");
                continue;
            }

            // if we get here the clusterId either:
            // 1) is inactive
            // 2) needs recovery
            if (doc.isRecoveryNeeded(now)) {
                if (mId.equals(machineId) && iId.equals(instanceId)) {
                    // this id matches our environment and has an expired lease
                    // use it after a successful recovery
                    if (!recoveryHandler.recover(id)) {
                        reuseFailureReason = reject(id,
                                "needs recovery and was unable to perform it myself");
                        continue;
                    }
                } else {
                    // a different machine or instance
                    reuseFailureReason = reject(id,
                            "needs recovery and machineId/instanceId do not match: " +
                                    mId + "/" + iId + " != " + machineId + "/" + instanceId);
                    continue;
                }
            }

            // if we get here the cluster node entry is inactive. if recovery
            // was needed, then it was successful

            // create a candidate. those with matching machine and instance id
            // are preferred, then the one with the lowest clusterId.
            candidates.add(new ClusterNodeInfo(id, store, mId, iId, false, invisible));
            startTimes.put(id, doc.getStartTime());
        }

        if (candidates.isEmpty()) {
            // No usable existing entry found
            int clusterNodeId;
            if (configuredClusterId != 0) {
                if (alreadyExistingConfigured != null) {
                    throw new DocumentStoreException(
                            "Configured cluster node id " + configuredClusterId + " already in use: " + reuseFailureReason);
                }
                clusterNodeId = configuredClusterId;
            } else {
                clusterNodeId = maxId + 1;
            }
            // No usable existing entry found so create a new entry
            candidates.add(new ClusterNodeInfo(clusterNodeId, store, machineId, instanceId, true, invisible));
        }

        // use the best candidate
        ClusterNodeInfo info = candidates.first();
        // and replace with an info matching the current machine and instance id
        info = new ClusterNodeInfo(info.id, store, machineId, instanceId, info.newEntry, invisible);
        return new AbstractMap.SimpleImmutableEntry<>(info, startTimes.get(info.getId()));
    }

    private static void logClusterIdAcquired(ClusterNodeInfo clusterNode,
                                             ClusterNodeInfoDocument before) {
        String type = clusterNode.newEntry ? "new" : "existing";
        type = clusterNode.invisible ? (type + " (invisible)") : type;

        String machineInfo = clusterNode.machineId;
        String instanceInfo = clusterNode.instanceId;
        if (before != null) {
            // machineId or instanceId may have changed
            String beforeMachineId = String.valueOf(before.get(MACHINE_ID_KEY));
            String beforeInstanceId = String.valueOf(before.get(INSTANCE_ID_KEY));
            if (!clusterNode.machineId.equals(beforeMachineId)) {
                machineInfo = "(changed) " + beforeMachineId + " -> " + machineInfo;
            }
            if (!clusterNode.instanceId.equals(beforeInstanceId)) {
                instanceInfo = "(changed) " + beforeInstanceId + " -> " + instanceInfo;
            }
        }
        LOG.info("Acquired ({}) clusterId {}. MachineId {}, InstanceId {}",
                type, clusterNode.getId(), machineInfo, instanceInfo);

    }

    private static String reject(int id, String reason) {
        LOG.debug("Cannot acquire {}: {}", id, reason);
        return reason;
    }

    private static boolean waitForLeaseExpiry(DocumentStore store, ClusterNodeInfoDocument cdoc, long leaseEnd, String machineId,
            String instanceId) {
        String key = cdoc.getId();
        LOG.info("Found an existing possibly active cluster node info (" + key + ") for this instance: " + machineId + "/"
                + instanceId + ", will try use it.");

        // wait until lease expiry plus 2s
        long waitUntil = leaseEnd + 2000;

        while (getCurrentTime() < waitUntil) {
            LOG.info("Waiting for cluster node " + key + "'s lease to expire: " + (waitUntil - getCurrentTime()) / 1000 + "s left");

            try {
                clock.waitUntil(getCurrentTime() + 5000);
            } catch (InterruptedException e) {
                // ignored
            }

            try {
                // check state of cluster node info
                ClusterNodeInfoDocument reread = store.find(Collection.CLUSTER_NODES, key);
                if (reread == null) {
                    LOG.info("Cluster node info " + key + ": gone; continuing.");
                    return true;
                } else {
                    Long newLeaseEnd = (Long) reread.get(LEASE_END_KEY);
                    if (newLeaseEnd == null) {
                        LOG.info("Cluster node " + key + ": lease end information missing, aborting.");
                        return false;
                    } else {
                        if (newLeaseEnd != leaseEnd) {
                            LOG.info("Cluster node " + key + " seems to be still active (lease end changed from " + leaseEnd
                                    + " to " + newLeaseEnd + ", will not try to use it.");
                            return false;
                        }
                    }
                }
            } catch (DocumentStoreException ex) {
                LOG.info("Error reading cluster node info for key " + key, ex);
                return false;
            }
        }
        return true;
    }

    /**
     * Checks if the lease for this cluster node is still valid, otherwise
     * throws a {@link DocumentStoreException}. Depending on the
     * {@link LeaseCheckMode} this method will not throw the
     * exception immediately when the lease expires. If the mode is set to
     * {@link LeaseCheckMode#LENIENT}, then this method will give the lease
     * update thread a last chance of 5 seconds to renew it. This allows the
     * DocumentNodeStore to recover from an expired lease caused by a system
     * put to sleep or a JVM in debug mode.
     *
     * @throws DocumentStoreException if the lease expired.
     */
    public void performLeaseCheck() throws DocumentStoreException {
        if (leaseCheckMode == LeaseCheckMode.DISABLED) {
            // if leaseCheckDisabled is set we never do the check, so return fast
            return;
        }
        if (leaseCheckFailed) {
            // unsynchronized access to leaseCheckFailed is fine
            // since it only ever changes from false to true once
            // and should the current thread read it erroneously
            // as false here, it would further down find out that
            // the lease has indeed still expired and then
            // go into the synchronized.
            // (note that once a lease check failed it would not
            // be updated again, ever, as guaranteed by checking
            // for leaseCheckFailed in renewLease() )
            throw leaseExpired(LEASE_CHECK_FAILED_MSG + " (since " + asISO8601(leaseEndTime) + ")", true);
        }
        long now = getCurrentTime();
        // OAK-3238 put the barrier 1/3 of 60sec=20sec before the end
        // OAK-3398 keeps this the same but uses an explicit leaseFailureMargin for this
        if (!isLeaseExpired(now)) {
            // then all is good
            return;
        }
        // synchronized: we need to guard leaseCheckFailed in order to ensure
        //               that it is only set by 1 thread - thus handleLeaseFailure
        //               is guaranteed to be only called once
        synchronized(this) {
            if (leaseCheckFailed) {
                // someone else won and marked leaseCheckFailed - so we only log/throw
                throw leaseExpired(LEASE_CHECK_FAILED_MSG + " (since " + asISO8601(leaseEndTime) + ")", true);
            }
            // only retry in lenient mode, fail immediately in strict mode
            final int maxRetries = leaseCheckMode == LeaseCheckMode.STRICT ?
                    0 : MAX_RETRY_SLEEPS_BEFORE_LEASE_FAILURE;
            for (int i = 0; i < maxRetries; i++) {
                now = getCurrentTime();
                if (!isLeaseExpired(now)) {
                    // if lease is OK here, then there was a race
                    // between performLeaseCheck and renewLease()
                    // where the winner was: renewLease().
                    // so: luckily we can continue here
                    return;
                }
                // OAK-3399 : in case of running into the leaseFailureMargin
                // (shortly, 20sec, before the lease times out), we're now doing
                // a short retry loop of 1sec sleeps (default 5x1sec=5sec),
                // to give this instance 'one last chance' before we have to
                // declare the lease as failed.
                // This sort of retry loop would allow situations such as
                // when running a single-node cluster and interrupting/pausing
                // the process temporarily: in this case when waking up, the
                // lease might momentarily be timed out, but the lease would
                // still be 'updateable' and that would happen pretty soon
                // after waking up. So in that case, doing these retry-sleeps
                // would help.
                // in most other cases where the local instance is not doing
                // lease updates due to 'GC-death' or 'lease-thread-crashed'
                // or the like, it would not help. But it would also not hurt
                // as the margin is 20sec and we're just reducing it by 5sec
                // (in the un-paused case)
                try {
                    long difference = leaseEndTime - now;
                    long waitForMs = 1000;

                    String detail = difference >= 0
                            ? String.format("lease within %dms of failing (%dms precisely)", leaseFailureMargin, difference)
                            : String.format("already past lease end (%dms precisely)", -1 * difference);
                    String retries = String.format("waiting %dms to retry (up to another %d times...)",
                            waitForMs, maxRetries - 1 - i);
                    LOG.info("performLeaseCheck: " + detail + " - " + retries);

                    // directly use this to sleep on - to allow renewLease() to work
                    wait(waitForMs);
                } catch (InterruptedException e) {
                    LOG.warn("performLeaseCheck: got interrupted - giving up: " + e, e);
                    break;
                }
            }
            if (leaseCheckFailed) {
                // someone else won and marked leaseCheckFailed - so we only log/throw
                throw leaseExpired(LEASE_CHECK_FAILED_MSG + " (since " + asISO8601(leaseEndTime) + ")", true);
            }
            leaseCheckFailed = true; // make sure only one thread 'wins', ie goes any further
        }

        String template = "%s (mode: %s, leaseEndTime: %d (%s), leaseTime: %d, leaseFailureMargin: %d, "
                + "lease check end time (leaseEndTime - leaseFailureMargin): %d (%s), now: %d (%s), remaining: %d)"
                + " Need to stop oak-store-document/DocumentNodeStoreService.";
        String errorMsg = String.format(template, LEASE_CHECK_FAILED_MSG, leaseCheckMode.name(), leaseEndTime,
                asISO8601(leaseEndTime), leaseTime, leaseFailureMargin, leaseEndTime - leaseFailureMargin,
                asISO8601(leaseEndTime - leaseFailureMargin), now, asISO8601(now), (leaseEndTime - leaseFailureMargin) - now);

        LOG.error(errorMsg);
        handleLeaseFailure(errorMsg);
    }

    /**
     * Returns {@code true} if the lease for this cluster node info should be
     * considered expired given the current {@code time}. This method takes
     * {@link #leaseFailureMargin} into account and will return {@code true}
     * even before the passed {@code time} is beyond the {@link #leaseEndTime}.
     *
     * @param time the current time to check against the lease end.
     * @return {@code true} if the lease is considered expired, {@code false}
     *         otherwise.
     */
    private boolean isLeaseExpired(long time) {
        return time >= (leaseEndTime - leaseFailureMargin);
    }

    private void handleLeaseFailure(final String errorMsg) {
        // OAK-3397 : unlike previously, when the lease check fails we should not
        // do a hard System exit here but rather stop the oak-store-document bundle
        // (or if that fails just deactivate DocumentNodeStore) - with the
        // goals to prevent this instance to continue to operate
        // give that a lease failure is a strong indicator of a faulty
        // instance - and to stop the background threads of DocumentNodeStore,
        // specifically the BackgroundLeaseUpdate and the BackgroundOperation.

        // actual stopping should be done in a separate thread, so:
        if (leaseFailureHandler!=null) {
            final Runnable r = new Runnable() {

                @Override
                public void run() {
                    if (leaseFailureHandler!=null) {
                        leaseFailureHandler.handleLeaseFailure();
                    }
                }
            };
            final Thread th = new Thread(r, "LeaseFailureHandler-Thread");
            th.setDaemon(true);
            th.start();
        }

        throw leaseExpired(errorMsg, false);
    }

    /**
     * Renew the cluster id lease. This method needs to be called once in a while,
     * to ensure the same cluster id is not re-used by a different instance.
     * The lease is only renewed after 'leaseUpdateInterval' millis
     * since last lease update - default being every 10 sec (this used to be 30sec).
     * <p>
     * This method will not fail immediately with a DocumentStoreException if
     * the lease expired. It will still try to renew the lease and only fail if
     * {@link #performLeaseCheck()} decided the lease expired or another cluster
     * node initiated recover for this node.
     *
     * @return {@code true} if the lease was renewed; {@code false} otherwise.
     * @throws DocumentStoreException if the operation failed or the lease
     *          expired.
     */
    public boolean renewLease() throws DocumentStoreException {
        long now = getCurrentTime();

        if (LOG.isTraceEnabled()) {
            LOG.trace("renewLease - leaseEndTime: " + leaseEndTime + ", leaseTime: " + leaseTime + ", leaseUpdateInterval: " + leaseUpdateInterval);
        }

        if (now < leaseEndTime - leaseTime + leaseUpdateInterval) {
            // no need to renew the lease - it is still within 'leaseUpdateInterval'
            return false;
        }
        // lease requires renewal

        long updatedLeaseEndTime;
        synchronized(this) {
            // this is synchronized since access to leaseCheckFailed and leaseEndTime
            // are both normally synchronized to propagate values between renewLease()
            // and performLeaseCheck().
            // (there are unsynchronized accesses to both of these as well - however
            // they are both double-checked - and with both reading a stale value is thus OK)

            if (leaseCheckFailed) {
                // prevent lease renewal after it failed
                throw leaseExpired(LEASE_CHECK_FAILED_MSG + " (since " + asISO8601(leaseEndTime) + ")", true);
            }
            // synchronized could have delayed the 'now', so
            // set it again..
            now = getCurrentTime();
            updatedLeaseEndTime = now + leaseTime;
        }

        if (leaseCheckMode == LeaseCheckMode.STRICT) {
            // check whether the lease is still valid and can be renewed
            if (isLeaseExpired(now)) {
                synchronized (this) {
                    if (leaseCheckFailed) {
                        // some other thread already noticed and calls failure handler
                        throw leaseExpired(LEASE_CHECK_FAILED_MSG + " (since " + asISO8601(leaseEndTime) + ")", true);
                    }
                    // current thread calls failure handler
                    // outside synchronized block
                    leaseCheckFailed = true;
                }
                String template = "%s (mode: %s, leaseEndTime: %d (%s), leaseTime: %d, leaseFailureMargin: %d, "
                        + "lease check end time (leaseEndTime - leaseFailureMargin): %d (%s), now: %d (%s), remaining: %d)"
                        + " Need to stop oak-store-document/DocumentNodeStoreService.";
                String errorMsg = String.format(template, LEASE_CHECK_FAILED_MSG, leaseCheckMode.name(), leaseEndTime,
                        asISO8601(leaseEndTime), leaseTime, leaseFailureMargin, leaseEndTime - leaseFailureMargin,
                        asISO8601(leaseEndTime - leaseFailureMargin), now, asISO8601(now), (leaseEndTime - leaseFailureMargin) - now);
                LOG.error(errorMsg);
                handleLeaseFailure(errorMsg);
                // should never be reached: handleLeaseFailure throws a DocumentStoreException
                return false;
            }
        }

        UpdateOp update = new UpdateOp("" + id, false);
        update.set(LEASE_END_KEY, updatedLeaseEndTime);

        if (leaseCheckMode != LeaseCheckMode.DISABLED) {
            // if leaseCheckDisabled, then we just update the lease without
            // checking
            // OAK-3398:
            // if we renewed the lease ever with this instance/ClusterNodeInfo
            // (which is the normal case.. except for startup),
            // then we can now make an assertion that the lease is unchanged
            // and the incremental update must only succeed if no-one else
            // did a recover/inactivation in the meantime
            // make three assertions: the leaseEnd must match ..
            update.equals(LEASE_END_KEY, null, previousLeaseEndTime);
            // plus it must still be active ..
            update.equals(STATE, null, ACTIVE.name());
            // plus it must not have a recovery lock on it
            update.notEquals(REV_RECOVERY_LOCK, ACQUIRED.name());
            // @TODO: to make it 100% failure proof we could introduce
            // yet another field to clusterNodes: a runtimeId that we
            // create (UUID) at startup each time - and against that
            // we could also check here - but that goes a bit far IMO
        }

        if (LOG.isDebugEnabled()) {
            LOG.debug("Renewing lease for cluster id " + id + " with UpdateOp " + update);
        }
        Stopwatch sw = Stopwatch.createStarted();
        DocumentStoreException dse;
        Object result = null;
        try {
            ClusterNodeInfoDocument doc = store.findAndUpdate(Collection.CLUSTER_NODES, update);
            result = doc;

            if (doc == null) { // should not occur when leaseCheckDisabled
                // OAK-3398 : someone else either started recovering or is already through with that.
                // in both cases the local instance lost the lease-update-game - and hence
                // should behave and must consider itself as 'lease failed'

                synchronized(this) {
                    if (leaseCheckFailed) {
                        // somehow the instance figured out otherwise that the
                        // lease check failed - so we don't have to too - so we just log/throw
                        throw leaseExpired(LEASE_CHECK_FAILED_MSG, true);
                    }
                    leaseCheckFailed = true; // make sure only one thread 'wins', ie goes any further
                }

                String errorMsg = LEASE_CHECK_FAILED_MSG
                        + " (Could not update lease anymore, someone else in the cluster "
                        + "must have noticed this instance' slowness already. "
                        + "Going to invoke leaseFailureHandler!)";

                // try to add more diagnostics
                try {
                    ClusterNodeInfoDocument current = store.find(Collection.CLUSTER_NODES, "" + id);
                    if (current != null) {
                        Object leaseEnd = current.get(LEASE_END_KEY);
                        Object recoveryLock = current.get(REV_RECOVERY_LOCK);
                        Object recoveryBy = current.get(REV_RECOVERY_BY);
                        Object cnState = current.get(STATE);
                        errorMsg += " (leaseEnd: " + leaseEnd + " (expected: " + leaseEndTime + ")" +
                                ", (state: " + cnState + " (expected: " + ACTIVE.name() + ")" +
                                ", recoveryLock: " + recoveryLock +
                                ", recoveryBy: " + recoveryBy + ")";
                    }
                } catch (DocumentStoreException ex) {
                    LOG.error("trying to read ClusterNodeInfo for cluster id " + id, ex);
                }

                LOG.error(errorMsg);

                handleLeaseFailure(errorMsg);
                // should never be reached: handleLeaseFailure throws a DocumentStoreException
                return false;
            }
            leaseEndTime = updatedLeaseEndTime;
            previousLeaseEndTime = leaseEndTime; // store previousLeaseEndTime for reference for next time
            String mode = (String) doc.get(READ_WRITE_MODE_KEY);
            if (mode != null && !mode.equals(readWriteMode)) {
                readWriteMode = mode;
                store.setReadWriteMode(mode);
            }
            return true;
        } catch (DocumentStoreException e) {
            dse = e;
            result = e.toString();
        } finally {
            sw.stop();
            String msg = "Lease renewal for cluster id {} took {}, resulted in: {}";
            if (sw.elapsed(TimeUnit.SECONDS) > 10) {
                LOG.warn(msg, id, sw, result);
            } else if (sw.elapsed(TimeUnit.SECONDS) > 1) {
                LOG.info(msg, id, sw, result);
            } else if (LOG.isDebugEnabled()) {
                LOG.debug(msg, id, sw, result);
            }
        }
        // if we get here, the update failed with an exception, try to read the
        // current cluster node info document and update leaseEndTime &
        // previousLeaseEndTime accordingly until leaseEndTime is reached
        while (getCurrentTime() < updatedLeaseEndTime) {
            synchronized (this) {
                if (leaseCheckFailed) {
                    // no need to read from store, lease check already failed
                    break;
                }
            }
            long t1 = clock.getTime();
            ClusterNodeInfoDocument doc;
            try {
                doc = store.find(Collection.CLUSTER_NODES, String.valueOf(id));
            } catch (DocumentStoreException e) {
                LOG.warn("Reading ClusterNodeInfoDocument for id " + id + " failed", e);
                // do not retry more than once a second
                try {
                    clock.waitUntil(t1 + 1000);
                } catch (InterruptedException iex) {
                    // ignore
                }
                continue;
            }
            if (doc != null) {
                if (!doc.isActive()) {
                    LOG.warn("ClusterNodeInfoDocument for id {} is not active " +
                            "anymore. {}", id, doc);
                    // break here and let the next lease update attempt fail
                    break;
                } else if (doc.getLeaseEndTime() == previousLeaseEndTime
                        || doc.getLeaseEndTime() == updatedLeaseEndTime) {
                    // set lease end times to current values
                    previousLeaseEndTime = doc.getLeaseEndTime();
                    leaseEndTime = doc.getLeaseEndTime();
                    break;
                } else {
                    // leaseEndTime is neither the previous nor the new value
                    // another cluster node must have updated the leaseEndTime
                    // break here and let the next lease update attempt fail
                    break;
                }
            } else {
                LOG.warn("ClusterNodeInfoDocument for id {} does not exist anymore", id);
                break;
            }
        }
        throw dse;
    }
    
    /**
     * Update the cluster node info.
     * 
     * @param info the map of changes
     */
    public void setInfo(Map<String, String> info) {
        // synchronized, because renewLease is also synchronized 
        synchronized(this) {
            UpdateOp update = new UpdateOp("" + id, false);
            for(Entry<String, String> e : info.entrySet()) {
                update.set(e.getKey(), e.getValue());
            }
            store.findAndUpdate(Collection.CLUSTER_NODES, update);
        }
    }

    /** for testing purpose only, not to be changed at runtime! */
    void setLeaseTime(long leaseTime) {
        this.leaseTime = leaseTime;
    }

    /** for testing purpose only, not to be changed at runtime! */
    void setLeaseUpdateInterval(long leaseUpdateInterval) {
        this.leaseUpdateInterval = leaseUpdateInterval;
    }

    public long getLeaseTime() {
        return leaseTime;
    }

    public long getLeaseEndTime() {
        return leaseEndTime;
    }

    public void setLeaseFailureHandler(LeaseFailureHandler leaseFailureHandler) {
        this.leaseFailureHandler = leaseFailureHandler;
    }

    public void dispose() {
        synchronized(this) {
            if (leaseCheckFailed) {
                LOG.warn("dispose: lease check failed, thus not marking instance as cleanly shut down.");
                return;
            }
        }
        UpdateOp update = new UpdateOp("" + id, true);
        update.set(LEASE_END_KEY, null);
        update.set(STATE, null);
        update.set(INVISIBLE, false);
        store.createOrUpdate(Collection.CLUSTER_NODES, update);
        state = NONE;
    }

    @Override
    public String toString() {
        return "id: " + id + ",\n" +
                "startTime: " + startTime + ",\n" +
                "machineId: " + machineId + ",\n" +
                "instanceId: " + instanceId + ",\n" +
                "pid: " + PROCESS_ID + ",\n" +
                "uuid: " + uuid + ",\n" +
                "readWriteMode: " + readWriteMode + ",\n" +
                "leaseCheckMode: " + leaseCheckMode.name() + ",\n" +
                "state: " + state + ",\n" +
                "oakVersion: " + OAK_VERSION + ",\n" +
                "formatVersion: " + DocumentNodeStore.VERSION + ",\n" +
                "invisible: " + invisible;
    }

    /**
     * Specify a custom clock to be used for determining current time.
     *
     * <b>Only Used For Testing</b>
     */
    static void setClock(Clock c) {
        checkNotNull(c);
        clock = c;
    }

    /**
     * Resets the clock to the default
     */
    static void resetClockToDefault() {
        clock = Clock.SIMPLE;
    }

    private static long getProcessId() {
        try {
            String name = ManagementFactory.getRuntimeMXBean().getName();
            return Long.parseLong(name.substring(0, name.indexOf('@')));
        } catch (Exception e) {
            LOG.warn("Could not get process id", e);
            return 0;
        }
    }

    /*
     * Allow external override of hardware address. The special value "(none)"
     * indicates that a situation where no hardware address is available is to
     * be simulated.
     */
    private static String getHWAFromSystemProperty() {
        return SystemPropertySupplier.create(ClusterNodeInfo.class.getName() + ".HWADDRESS", "").loggingTo(LOG)
                .logSuccessAs(Level.DEBUG)
                .formatSetMessage(
                        (name, value) -> String.format("obtaining hardware address from system variable %s: %s", name, value))
                .get();
    }

    /**
     * Calculate the unique machine id. This usually is the lowest MAC address
     * if available. As an alternative, a randomly generated UUID is used.
     *
     * @return the unique id
     */
    private static String getHardwareMachineId() {
        Exception exception = null;
        try {
            ArrayList<String> macAddresses = new ArrayList<String>();
            ArrayList<String> likelyVirtualMacAddresses = new ArrayList<String>();
            ArrayList<String> otherAddresses = new ArrayList<String>();
            String hwaFromSysProp = getHWAFromSystemProperty();
            if ("".equals(hwaFromSysProp)) {
                Enumeration<NetworkInterface> e = NetworkInterface
                        .getNetworkInterfaces();
                while (e.hasMoreElements()) {
                    NetworkInterface ni = e.nextElement();
                    try {
                        byte[] hwa = ni.getHardwareAddress();
                        // empty addresses have been seen on loopback devices
                        if (hwa != null && hwa.length != 0) {
                            String str = StringUtils.convertBytesToHex(hwa);
                            if (hwa.length == 6) {
                                // likely a MAC address
                                String displayName = ni.getDisplayName().toLowerCase(Locale.ENGLISH);
                                // de-prioritize addresses that are likely to be virtual (see OAK-3885 and OAK-8935)
                                boolean looksVirtual = displayName.indexOf("virtual") >= 0 || displayName.indexOf("vpn") >= 0
                                        || displayName.indexOf("docker") >= 0;
                                if (!looksVirtual) {
                                    macAddresses.add(str);
                                } else {
                                    likelyVirtualMacAddresses.add(str);
                                }
                            } else {
                                otherAddresses.add(str);
                            }
                        }
                    } catch (Exception e2) {
                        exception = e2;
                    }
                }
            }
            else {
                if (!"(none)".equals(hwaFromSysProp)) {
                    if (hwaFromSysProp.length() == 12) {
                        // assume 12 hex digits are a mac address
                        macAddresses.add(hwaFromSysProp);
                    } else {
                        otherAddresses.add(hwaFromSysProp);
                    }
                }
            }

            if (LOG.isDebugEnabled()) {
                LOG.debug("getMachineId(): discovered addresses: {} {} {}", macAddresses, likelyVirtualMacAddresses,
                        otherAddresses);
            }

            if (macAddresses.size() > 0) {
                // use the lowest MAC value, such that if the order changes,
                // the same one is used
                Collections.sort(macAddresses);
                return "mac:" + macAddresses.get(0);
            } else if (likelyVirtualMacAddresses.size() > 0) {
                Collections.sort(likelyVirtualMacAddresses);
                return "mac:" + likelyVirtualMacAddresses.get(0);
            } else if (otherAddresses.size() > 0) {
                // try the lowest "other" address
                Collections.sort(otherAddresses);
                return "hwa:" + otherAddresses.get(0);
            }
        } catch (Exception e) {
            exception = e;
        }
        if (exception != null) {
            LOG.warn("Error getting the machine id; using a UUID", exception);
        }
        return RANDOM_PREFIX + UUID.randomUUID().toString();
    }

    private static long getCurrentTime() {
        return clock.getTime();
    }

    private static DocumentStoreException leaseExpired(String msg, boolean log) {
        if (log) {
            LOG.error(msg);
        }
        return new DocumentStoreException(msg);
    }

    private static String asISO8601(long ms) {
        return DateTimeFormatter.ISO_INSTANT.format(Instant.ofEpochMilli(ms));
   }
}