/*
 * 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 java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import javax.jcr.Value;

import org.apache.jackrabbit.commons.SimpleValueFactory;
import org.apache.jackrabbit.oak.api.Descriptors;
import org.apache.jackrabbit.oak.osgi.OsgiWhiteboard;
import org.apache.jackrabbit.oak.spi.cluster.ClusterRepositoryInfo;
import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
import org.apache.jackrabbit.oak.spi.commit.Observer;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.jetbrains.annotations.NotNull;
import org.osgi.framework.Version;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.component.annotations.Activate;
import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.Deactivate;
import org.osgi.service.component.annotations.Reference;
import org.osgi.service.component.annotations.ReferenceCardinality;
import org.osgi.service.component.annotations.ReferencePolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The DocumentDiscoveryLiteService is taking care of providing a repository
 * descriptor that contains the current cluster-view details.
 * <p>
 * The clusterView is provided via a repository descriptor (see
 * OAK_DISCOVERYLITE_CLUSTERVIEW)
 * <p>
 * The cluster-view lists all instances (ever) known in the cluster in one of
 * the following states:
 * <ul>
 * <li>active: the instance is currently running and has an up-to-date lease
 * </li>
 * <li>deactivating: the instance failed to update the lease recently thus a
 * recovery is happening - or it has just finished and the local instance is yet
 * to do a backgroundRead before it has finished reading the crashed/shutdown
 * instance's last changes</li>
 * <li>inactive: the instance is currently not running and all its changes have
 * been seen by the local instance</li>
 * </ul>
 * <p>
 * Additionally, the cluster-view is assigned a monotonically increasing
 * sequence number to. This sequence number is persisted, thus all instances in
 * the cluster will show the same sequence number for a particular cluster-view
 * in time.
 * <p>
 * Note that the 'deactivating' state might be hiding some complexity that is
 * deliberately not shown: for the documentNS the state 'deactivating' consists
 * of two substates: 'recovering' as in _lastRevs are updated, and 'backlog
 * processing' for a pending backgroundRead to get the latest head state of a
 * crashed/shutdown instance. So when an instance is in 'deactivating' state, it
 * is not indicated via the cluster-view whether it is recovering or has backlog
 * to process. However, the fact that an instance has to yet do a backgroundRead
 * to get changes is a per-instance story: other instances might already have
 * done the backgroundRead and thus no longer have a backlog for the instance(s)
 * that left. Even though 'deactivating' therefore is dependent on the instance
 * you get the information from, the cluster-view must have a sequence number
 * that uniquely identifies it in the cluster. These two constraints conflict.
 * As a simple solution to handle this case nevertheless, the 'final' flag has
 * been introduced: the cluster-view has this flag 'final' set to true when the
 * view is final and nothing will be changed in this sequence number anymore. If
 * the 'final' flag is false however it indicates that the cluster-view with
 * this particular sequence number might still experience a change (more
 * concretely: the deactivating instances might change). Note that there
 * alternatives to this 'final' flag have been discussed, such as using
 * vector-counters, but there was no obvious gain achieve using an alternative
 * approach.
 * <p>
 * In other words: whenever the 'final' flag is false, the view must be
 * interpreted as 'in flux' wrt the deactivating/inactive instances and any
 * action that depends on stable deactivating/inactive instances must not yet be
 * done until the 'final' flag becomes true.
 * <p>
 * Underneath, the DocumentDiscoveryLiteService uses the clusterNodes collection
 * to derive the clusterView, which it stores in the settings collection.
 * Whenever it updates the clusterView it increments the sequence number by 1.
 * <p>
 * While this new 'clusterView' document in the settings collection sounds like
 * redundant data (since it is just derived from the clusterNodes), it actually
 * is not. By persisting the clusterView it becomes the new source of truth wrt
 * what the clusterView looks like. And no two instances in the same cluster can
 * make different conclusions based eg on different clocks they have or based on
 * reading the clusterNodes in a slightly different moment etc. Also, the
 * clusterView allows to store a the sequence number
 * (which allows the instances to make reference to the same clusterView, and be
 * able to simply detect whether anything has changed)
 * <p>
 * Prerequisites that the clusterView mechanism is stable:
 * <ul>
 * <li>the machine clocks are reasonably in sync - that is, they should be off
 * by magnitudes less than the lease updateFrequency/timeout</li>
 * <li>the write-delays from any instance to the mongo server where the
 * clusterNodes and settings collections are stored should be very fast - at
 * least orders of magnitudes lower again than the lease timeout</li>
 * <li>when this instance notices that others have kicked it out of the
 * clusterView (which it can find out when either its clusterNodes document is
 * set to recovering or it is not in the clusterView anymore, although it just
 * was - ie not just because of a fresh start), then this instance must step
 * back gracefully. The exact definition is to be applied elsewhere - but it
 * should include: stopping to update its own lease, waiting for the view to
 * have stabilized - waiting for recovery of its own instance by the remaining
 * instances in the cluster to have finished - and then probably waiting for
 * another gracePeriod until it might rejoin the cluster. In between, any commit
 * should fail with BannedFromClusterException</li>
 * </ul>
 * 
 * @see #OAK_DISCOVERYLITE_CLUSTERVIEW
 */
@Component(
        name = DocumentDiscoveryLiteService.COMPONENT_NAME,
        immediate = true,
        service = { DocumentDiscoveryLiteService.class, Observer.class })
public class DocumentDiscoveryLiteService implements ClusterStateChangeListener, Observer {

    static final String COMPONENT_NAME = "org.apache.jackrabbit.oak.plugins.document.DocumentDiscoveryLiteService";

    /**
     * Name of the repository descriptor via which the clusterView is published
     * - which is the raison d'etre of the DocumentDiscoveryLiteService
     **/
    public static final String OAK_DISCOVERYLITE_CLUSTERVIEW = "oak.discoverylite.clusterview";

    private static final Logger logger = LoggerFactory.getLogger(DocumentDiscoveryLiteService.class);

    /** describes the reason why the BackgroundWorker should be woken up **/
    private static enum WakeupReason {
        CLUSTER_STATE_CHANGED, BACKGROUND_READ_FINISHED
    }

    /**
     * The BackgroundWorker is taking care of regularly invoking checkView -
     * which in turn will detect if anything changed
     **/
    private class BackgroundWorker implements Runnable {

        final Random random = new Random();

        boolean stopped = false;

        private void stop() {
            logger.trace("stop: start");
            synchronized (BackgroundWorker.this) {
                stopped = true;
            }
            logger.trace("stop: end");
        }

        @Override
        public void run() {
            logger.info("BackgroundWorker.run: start");
            try {
                doRun();
            } finally {
                logger.info("BackgroundWorker.run: end {finally}");
            }
        }

        private void doRun() {
            while (!stopped) {
                try {
                    logger.trace("BackgroundWorker.doRun: going to call checkView");
                    boolean shortSleep = checkView();
                    logger.trace("BackgroundWorker.doRun: checkView terminated with {} (=shortSleep)", shortSleep);
                    long sleepMillis = shortSleep ? (50 + random.nextInt(450)) : 5000;
                    logger.trace("BackgroundWorker.doRun: sleeping {}ms", sleepMillis);
                    synchronized (BackgroundWorker.this) {
                        if (stopped)
                            return;
                        BackgroundWorker.this.wait(sleepMillis);
                        if (stopped)
                            return;
                    }
                    logger.trace("BackgorundWorker.doRun: done sleeping, looping");
                } catch (Exception e) {
                    logger.error("doRun: got an exception: " + e, e);
                    try {
                        Thread.sleep(5000);
                    } catch (Exception e2) {
                        logger.error("doRun: got an exception while sleeping due to another exception: " + e2, e2);
                    }
                }
            }
        }

    }

    /** This provides the 'clusterView' repository descriptors **/
    private class DiscoveryLiteDescriptor implements Descriptors {

        final SimpleValueFactory factory = new SimpleValueFactory();

        @Override
        public String[] getKeys() {
            return new String[] { OAK_DISCOVERYLITE_CLUSTERVIEW };
        }

        @Override
        public boolean isStandardDescriptor(String key) {
            if (!OAK_DISCOVERYLITE_CLUSTERVIEW.equals(key)) {
                return false;
            }
            return true;
        }

        @Override
        public boolean isSingleValueDescriptor(String key) {
            if (!OAK_DISCOVERYLITE_CLUSTERVIEW.equals(key)) {
                return false;
            }
            return true;
        }

        @Override
        public Value getValue(String key) {
            if (!OAK_DISCOVERYLITE_CLUSTERVIEW.equals(key)) {
                return null;
            }
            return factory.createValue(getClusterViewAsDescriptorValue());
        }

        @Override
        public Value[] getValues(String key) {
            if (!OAK_DISCOVERYLITE_CLUSTERVIEW.equals(key)) {
                return null;
            }
            return new Value[] { getValue(key) };
        }

    }

    /** DocumentNodeStore's (hence local) clusterId **/
    private int clusterNodeId = -1;

    /**
     * the DocumentNodeStore - used to get the active/inactive cluster ids from
     **/
    private DocumentNodeStore documentNodeStore;

    /**
     * background job that periodically verifies and updates the clusterView
     **/
    private BackgroundWorker backgroundWorker;

    /** the ClusterViewDocument which was used in the last checkView run **/
    private ClusterViewDocument previousClusterViewDocument;

    /**
     * the ClusterView that was valid as a result of the previous checkView run
     **/
    private ClusterView previousClusterView;

    /**
     * kept volatile as this is frequently read in contentChanged which is
     * better kept unsynchronized as long as possible
     **/
    private volatile boolean hasInstancesWithBacklog;

    /**
     * Require a static reference to the NodeStore. Note that this implies the
     * service is only active for documentNS
     **/
    @Reference
    private DocumentNodeStore nodeStore;

    /**
     * inactive nodes that have been so for a while, ie they have no backlog
     * anymore, so no need to check for backlog every time
     **/
    private Set<Integer> longTimeInactives = new HashSet<Integer>();

    /**
     * returns the clusterView as a json value for it to be provided via the
     * repository descriptor
     **/
    private String getClusterViewAsDescriptorValue() {
        if (previousClusterView == null) {
            return null;
        } else {
            return previousClusterView.asDescriptorValue();
        }
    }

    /**
     * On activate the DocumentDiscoveryLiteService tries to start the
     * background job
     */
    @Activate
    public void activate(ComponentContext context) {
        logger.trace("activate: start");

        // set the ClusterStateChangeListener with the DocumentNodeStore
        this.documentNodeStore = (DocumentNodeStore) nodeStore;
        documentNodeStore.setClusterStateChangeListener(this);

        // retrieve the clusterId
        clusterNodeId = documentNodeStore.getClusterId();

        // start the background worker
        backgroundWorker = new BackgroundWorker();
        Thread th = new Thread(backgroundWorker, "DocumentDiscoveryLiteService-BackgroundWorker-[" + clusterNodeId + "]");
        th.setDaemon(true);
        th.start();

        // register the Descriptors - for Oak to pass it upwards
        if (context != null) {
            OsgiWhiteboard whiteboard = new OsgiWhiteboard(context.getBundleContext());
            whiteboard.register(Descriptors.class, new DiscoveryLiteDescriptor(), Collections.emptyMap());
        }
        logger.trace("activate: end");
    }

    /**
     * On deactivate the background job is stopped - if it was running at all
     **/
    @Deactivate
    protected void deactivate() {
        logger.trace("deactivate: deactivated");

        if (backgroundWorker != null) {
            backgroundWorker.stop();
            backgroundWorker = null;
        }
        logger.trace("deactivate: end");
    }

    /**
     * Checks if anything changed in the current view and updates the service
     * fields accordingly.
     * 
     * @return true if anything changed or is about to be changed (eg
     *         recovery/backlog), false if the view is stable
     */
    private boolean checkView() {
        logger.trace("checkView: start");
        List<ClusterNodeInfoDocument> allClusterNodes = ClusterNodeInfoDocument.all(documentNodeStore.getDocumentStore());

        Map<Integer, ClusterNodeInfoDocument> allNodeIds = new HashMap<Integer, ClusterNodeInfoDocument>();
        Map<Integer, ClusterNodeInfoDocument> activeNotTimedOutNodes = new HashMap<Integer, ClusterNodeInfoDocument>();
        Map<Integer, ClusterNodeInfoDocument> activeButTimedOutNodes = new HashMap<Integer, ClusterNodeInfoDocument>();
        Map<Integer, ClusterNodeInfoDocument> recoveringNodes = new HashMap<Integer, ClusterNodeInfoDocument>();
        Map<Integer, ClusterNodeInfoDocument> backlogNodes = new HashMap<Integer, ClusterNodeInfoDocument>();
        Map<Integer, ClusterNodeInfoDocument> inactiveNoBacklogNodes = new HashMap<Integer, ClusterNodeInfoDocument>();

        for (Iterator<ClusterNodeInfoDocument> it = allClusterNodes.iterator(); it.hasNext();) {
            ClusterNodeInfoDocument clusterNode = it.next();
            if (!clusterNode.isInvisible()) {
                allNodeIds.put(clusterNode.getClusterId(), clusterNode);
                if (clusterNode.isBeingRecovered()) {
                    recoveringNodes.put(clusterNode.getClusterId(), clusterNode);
                } else if (!clusterNode.isActive()) {
                    if (hasBacklog(clusterNode)) {
                        backlogNodes.put(clusterNode.getClusterId(), clusterNode);
                    } else {
                        inactiveNoBacklogNodes.put(clusterNode.getClusterId(), clusterNode);
                    }
                } else if (clusterNode.getLeaseEndTime() < System.currentTimeMillis()) {
                    activeButTimedOutNodes.put(clusterNode.getClusterId(), clusterNode);
                } else {
                    activeNotTimedOutNodes.put(clusterNode.getClusterId(), clusterNode);
                }
            }
        }

        // the current view should now consist of:
        // activeNotTimedOutNodes and activeButTimedOutNodes!
        // (reason for including the timedout: they will yet have to
        // switch to recovering or inactive - but we DONT KNOW yet.. that's
        // predicting the future - so so far we have to stick with
        // including them in the view)
        Map<Integer, ClusterNodeInfoDocument> allActives;
        allActives = new HashMap<Integer, ClusterNodeInfoDocument>(activeNotTimedOutNodes);
        allActives.putAll(activeButTimedOutNodes);

        // terminology:
        // 'inactivating' are nodes that are either 'recovering' or 'backlog'
        // ones
        // 'recovering' are nodes for which one node is doing the recover() of
        // lastRevs
        // 'backlog' ones are nodes that are no longer active, that have
        // finished the
        // recover() but for which a backgroundRead is still pending to read
        // the latest root changes.

        logger.debug(
                "checkView: active nodes: {}, timed out nodes: {}, recovering nodes: {}, backlog nodes: {}, inactive nodes: {}, total: {}, hence view nodes: {}",
                activeNotTimedOutNodes.size(), activeButTimedOutNodes.size(), recoveringNodes.size(), backlogNodes.size(),
                inactiveNoBacklogNodes.size(), allNodeIds.size(), allActives.size());

        ClusterViewDocument originalView = previousClusterViewDocument;
        ClusterViewDocument newView = doCheckView(allActives.keySet(), recoveringNodes.keySet(), backlogNodes.keySet(),
                inactiveNoBacklogNodes.keySet());
        if (newView == null) {
            logger.trace("checkView: end. newView: null");
            return true;
        }
        boolean newHasInstancesWithBacklog = recoveringNodes.size() > 0 || backlogNodes.size() > 0;
        boolean changed = originalView == null || (newView.getViewSeqNum() != originalView.getViewSeqNum())
                || (newHasInstancesWithBacklog != hasInstancesWithBacklog);
        logger.debug("checkView: viewFine: {}, changed: {}, originalView: {}, newView: {}", newView != null, changed, originalView,
                newView);

        if (longTimeInactives.addAll(inactiveNoBacklogNodes.keySet())) {
            logger.debug("checkView: updated longTimeInactives to {} (inactiveNoBacklogNodes: {})", longTimeInactives,
                    inactiveNoBacklogNodes);
        }

        if (changed) {
            String clusterId = ClusterRepositoryInfo.getOrCreateId(documentNodeStore);
            ClusterView v = ClusterView.fromDocument(clusterNodeId, clusterId, newView, backlogNodes.keySet());
            ClusterView previousView = previousClusterView;
            previousClusterView = v;
            hasInstancesWithBacklog = newHasInstancesWithBacklog;
            logger.info("checkView: view changed from: " + previousView + ", to: " + v + ", hasInstancesWithBacklog: "
                    + hasInstancesWithBacklog);
            return true;
        } else {
            logger.debug("checkView: no changes whatsoever, still at view: " + previousClusterView);
            return hasInstancesWithBacklog;
        }
    }

    private Revision getLastKnownRevision(int clusterNodeId) {
        String[] lastKnownRevisions = documentNodeStore.getMBean().getLastKnownRevisions();
        for (int i = 0; i < lastKnownRevisions.length; i++) {
            String aLastKnownRevisionStr = lastKnownRevisions[i];
            String[] split = aLastKnownRevisionStr.split("=");
            if (split.length == 2) {
                try {
                    Integer id = Integer.parseInt(split[0]);
                    if (id == clusterNodeId) {
                        final Revision lastKnownRev = Revision.fromString(split[1]);
                        logger.trace("getLastKnownRevision: end. clusterNode: {}, lastKnownRevision: {}", clusterNodeId,
                                lastKnownRev);
                        return lastKnownRev;
                    }
                } catch (NumberFormatException nfe) {
                    logger.warn("getLastKnownRevision: could not parse integer '" + split[0] + "': " + nfe, nfe);
                }
            } else {
                logger.warn("getLastKnownRevision: cannot parse lastKnownRevision: " + aLastKnownRevisionStr);
            }
        }
        logger.warn("getLastKnownRevision: no lastKnownRevision found for " + clusterNodeId);
        return null;
    }

    /** package access only for testing **/
    boolean hasBacklog(ClusterNodeInfoDocument clusterNode) {
        if (logger.isTraceEnabled()) {
            logger.trace("hasBacklog: start. clusterNodeId: {}", clusterNode.getClusterId());
        }
        Revision lastKnownRevision = getLastKnownRevision(clusterNode.getClusterId());
        if (lastKnownRevision == null) {
            logger.warn("hasBacklog: no lastKnownRevision found, hence cannot determine backlog for node "
                    + clusterNode.getClusterId());
            return false;
        }

        // The lastKnownRevision is what the local instance has last read/seen
        // from another instance.
        // This must be compared to what the other instance *actually* has
        // written as the very last thing.
        // Now the knowledge what the other instance has last written (after
        // recovery) would sit
        // in the root document - so that could in theory be used. But reading
        // the root document
        // would have to be done *uncached*. And that's quite a change to what
        // the original
        // idea was: that the root document would only be read every second, to
        // avoid contention.
        // So this 'what the other instance has last written' information is
        // retrieved via
        // a new, dedicated property in the clusterNodes collection: the
        // 'lastWrittenRootRev'.
        // The 'lastWrittenRootRev' is written by 'UnsavedModifications' during
        // backgroundUpdate
        // and retrieved here quite regularly (but it should not be a big deal,
        // as the
        // discovery-lite is the only one reading this field so frequently and
        // it does not
        // interfere with normal (jcr) nodes at all).
        String lastWrittenRootRevStr = clusterNode.getLastWrittenRootRev();
        if (lastWrittenRootRevStr == null) {
            boolean warn = false;
            Object oakVersion = clusterNode.get(ClusterNodeInfo.OAK_VERSION_KEY);
            if (oakVersion != null && (oakVersion instanceof String)) {
                try {
                    warn = versionPredates((String) oakVersion, "1.3.5");
                } catch (Exception e) {
                    logger.debug("hasBacklog: couldn't parse version " + oakVersion + " : " + e);
                    warn = true;
                }
            }
            if (warn) {
                logger.warn("hasBacklog: node has lastWrittenRootRev=null");
            } else {
                logger.debug("hasBacklog: node has lastWrittenRootRev=null");
            }
            return false;
        }
        Revision lastWrittenRootRev = Revision.fromString(lastWrittenRootRevStr);
        if (lastWrittenRootRev == null) {
            logger.warn("hasBacklog: node has no lastWrittenRootRev: " + clusterNode.getClusterId());
            return false;
        }

        boolean hasBacklog = Revision.getTimestampDifference(lastKnownRevision, lastWrittenRootRev) < 0;
        if (logger.isDebugEnabled()) {
            logger.debug("hasBacklog: clusterNodeId: {}, lastKnownRevision: {}, lastWrittenRootRev: {}, hasBacklog: {}",
                    clusterNode.getClusterId(), lastKnownRevision, lastWrittenRootRev, hasBacklog);
        }
        return hasBacklog;
    }

    static boolean versionPredates(String base, String compare) {
        Version one = Version.parseVersion(substSnapshotPrefix(compare));
        Version two = Version.parseVersion(substSnapshotPrefix(base));
        return two.compareTo(one) > 0;
    }

    private static String substSnapshotPrefix(String version) {
        String snapshot = "-SNAPSHOT";
        return version.endsWith(snapshot) ? version.substring(0, version.length() - snapshot.length()) + ".999999" : version;
    }

    private ClusterViewDocument doCheckView(final Set<Integer> activeNodes, final Set<Integer> recoveringNodes,
            final Set<Integer> backlogNodes, final Set<Integer> inactiveNodes) {
        logger.trace("doCheckView: start: activeNodes: {}, recoveringNodes: {}, backlogNodes: {}, inactiveNodes: {}", activeNodes,
                recoveringNodes, backlogNodes, inactiveNodes);

        Set<Integer> allInactives = new HashSet<Integer>();
        allInactives.addAll(inactiveNodes);
        allInactives.addAll(backlogNodes);

        if (activeNodes.size() == 0) {
            // then we have zero active nodes - that's nothing expected as that
            // includes our own node not to be active
            // hence handle with care - ie wait until we get an active node
            logger.warn("doCheckView: empty active ids. activeNodes:{}, recoveringNodes:{}, inactiveNodes:{}", activeNodes,
                    recoveringNodes, inactiveNodes);
            return null;
        }
        ClusterViewDocument newViewOrNull;
        try {
            newViewOrNull = ClusterViewDocument.readOrUpdate(documentNodeStore, activeNodes, recoveringNodes, allInactives);
        } catch (RuntimeException re) {
            logger.error("doCheckView: RuntimeException: re: " + re, re);
            return null;
        } catch (Error er) {
            logger.error("doCheckView: Error: er: " + er, er);
            return null;
        }
        logger.trace("doChckView: readOrUpdate result: {}", newViewOrNull);

        // and now for some verbose documentation and logging:
        if (newViewOrNull == null) {
            // then there was a concurrent update of the clusterView
            // and we should do some quick backoff sleeping
            logger.debug("doCheckView: newViewOrNull is null: " + newViewOrNull);
            return null;
        } else {
            // otherwise we now hold the newly valid view
            // it could be the same or different to the previous one, let's
            // check
            if (previousClusterViewDocument == null) {
                // oh ok, this is the very first one
                previousClusterViewDocument = newViewOrNull;
                logger.debug("doCheckView: end. first ever view: {}", newViewOrNull);
                return newViewOrNull;
            } else if (previousClusterViewDocument.getViewSeqNum() == newViewOrNull.getViewSeqNum()) {
                // that's the normal case: the viewId matches, nothing has
                // changed, we've already
                // processed the previousClusterView, so:
                logger.debug("doCheckView: end. seqNum did not change. view: {}", newViewOrNull);
                return newViewOrNull;
            } else {
                // otherwise the view has changed
                logger.info("doCheckView: view has changed from: {} to: {} - sending event...", previousClusterViewDocument,
                        newViewOrNull);
                previousClusterViewDocument = newViewOrNull;
                logger.debug("doCheckView: end. changed view: {}", newViewOrNull);
                return newViewOrNull;
            }
        }
    }

    @Override
    public void handleClusterStateChange() {
        // handleClusterStateChange is needed to learn about any state change in
        // the clusternodes
        // collection asap and being able to react on it - so this will wake up
        // the
        // backgroundWorker which in turn will - in a separate thread - check
        // the view
        // and send out events accordingly
        wakeupBackgroundWorker(WakeupReason.CLUSTER_STATE_CHANGED);
    }

    private void wakeupBackgroundWorker(WakeupReason wakeupReason) {
        final BackgroundWorker bw = backgroundWorker;
        if (bw != null) {
            // get a copy of this.hasInstancesWithBacklog for just the code-part
            // in this synchronized
            boolean hasInstancesWithBacklog = this.hasInstancesWithBacklog;

            if (wakeupReason == WakeupReason.BACKGROUND_READ_FINISHED) {
                // then only forward the notify if' hasInstancesWithBacklog'
                // ie, we have anything we could be waiting for - otherwise
                // we dont need to wakeup the background thread
                if (!hasInstancesWithBacklog) {
                    logger.trace(
                            "wakeupBackgroundWorker: not waking up backgroundWorker, as we do not have any instances with backlog");
                    return;
                }
            }
            logger.trace("wakeupBackgroundWorker: waking up backgroundWorker, reason: {} (hasInstancesWithBacklog: {})",
                    wakeupReason, hasInstancesWithBacklog);
            synchronized (bw) {
                bw.notifyAll();
            }
        }
    }

    /**
     * <p>
     * Additionally the DocumentDiscoveryLiteService must be notified when the
     * background-read has finished - as it could be waiting for a crashed
     * node's recovery to finish - which it can only do by checking the
     * lastKnownRevision of the crashed instance - and that check is best done
     * after the background read is just finished (it could optionally do that
     * just purely time based as well, but going via a listener is more timely,
     * that's why this approach has been chosen).
     */
    @Override
    public void contentChanged(@NotNull NodeState root,@NotNull CommitInfo info) {
        // contentChanged is only used to react as quickly as possible
        // when we have instances that have a 'backlog' - ie when instances
        // crashed
        // and are being recovered - then we must wait until the recovery is
        // finished
        // AND until the subsequent background read actually reads that
        // instance'
        // last changes. To catch that moment as quickly as possible,
        // this contentChanged is used.
        // Now from the above it also results that this only wakes up the
        // backgroundWorker if we have any pending 'backlogy instances'
        // otherwise this is a no-op
        if (info.isExternal()) {
            // then ignore this as this is likely an external change
            // note: it could be a compacted change, in which case we should
            // probably still process it - but we have a 5sec fallback
            // in the BackgroundWorker to handle that case too,
            // so:
            logger.trace("contentChanged: ignoring content change due to commit info belonging to external change");
            return;
        }
        logger.trace("contentChanged: handling content changed by waking up worker if necessary");
        wakeupBackgroundWorker(WakeupReason.BACKGROUND_READ_FINISHED);
    }

}