Rewrite and refactor core cluster controller state generation logic

Cluster controller will now generate the new cluster state on-demand in a "pure functional" way instead of conditionally patching a working state over time. This makes understanding (and changing) the state generation logic vastly easier than it previously was.

18 files changed, 2117 insertions, 296 deletions

diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterFixture.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterFixture.java
index aca26000931..3eda886e721 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterFixture.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterFixture.java
@@ -3,21 +3,18 @@ package com.yahoo.vespa.clustercontroller.core;
 
 import com.yahoo.vdslib.distribution.ConfiguredNode;
 import com.yahoo.vdslib.distribution.Distribution;
+import com.yahoo.vdslib.state.ClusterState;
 import com.yahoo.vdslib.state.Node;
 import com.yahoo.vdslib.state.NodeState;
 import com.yahoo.vdslib.state.NodeType;
 import com.yahoo.vdslib.state.State;
 import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler;
-import com.yahoo.vespa.clustercontroller.core.mocks.TestEventLog;
 import com.yahoo.vespa.clustercontroller.utils.util.NoMetricReporter;
-import com.yahoo.vespa.config.content.StorDistributionConfig;
 
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 
 import static org.mockito.Mockito.mock;
 
@@ -26,98 +23,163 @@ class ClusterFixture {
     public final Distribution distribution;
     public final FakeTimer timer;
     public final EventLogInterface eventLog;
-    public final SystemStateGenerator generator;
+    public final StateChangeHandler nodeStateChangeHandler;
+    public final ClusterStateGenerator.Params params = new ClusterStateGenerator.Params();
 
-    public ClusterFixture(ContentCluster cluster, Distribution distribution) {
+    ClusterFixture(ContentCluster cluster, Distribution distribution) {
         this.cluster = cluster;
         this.distribution = distribution;
         this.timer = new FakeTimer();
         this.eventLog = mock(EventLogInterface.class);
-        this.generator = createGeneratorForFixtureCluster();
+        this.nodeStateChangeHandler = createNodeStateChangeHandlerForCluster();
+        this.params.cluster(this.cluster);
     }
 
-    public SystemStateGenerator createGeneratorForFixtureCluster() {
+    StateChangeHandler createNodeStateChangeHandlerForCluster() {
         final int controllerIndex = 0;
         MetricUpdater metricUpdater = new MetricUpdater(new NoMetricReporter(), controllerIndex);
-        SystemStateGenerator generator = new SystemStateGenerator(timer, eventLog, metricUpdater);
-        generator.setNodes(cluster.clusterInfo());
-        generator.setDistribution(distribution);
-        return generator;
+        return new StateChangeHandler(timer, eventLog, metricUpdater);
     }
 
-    public void bringEntireClusterUp() {
+    ClusterFixture bringEntireClusterUp() {
         cluster.clusterInfo().getConfiguredNodes().forEach((idx, node) -> {
             reportStorageNodeState(idx, State.UP);
             reportDistributorNodeState(idx, State.UP);
         });
+        return this;
     }
 
-    public void reportStorageNodeState(final int index, State state) {
-        final Node node = new Node(NodeType.STORAGE, index);
-        final NodeState nodeState = new NodeState(NodeType.STORAGE, state);
-        nodeState.setDescription("mockdesc");
+    ClusterFixture markEntireClusterDown() {
+        cluster.clusterInfo().getConfiguredNodes().forEach((idx, node) -> {
+            reportStorageNodeState(idx, State.DOWN);
+            reportDistributorNodeState(idx, State.DOWN);
+        });
+        return this;
+    }
+
+    private void doReportNodeState(final Node node, final NodeState nodeState) {
+        final ClusterState stateBefore = rawGeneratedClusterState();
+
         NodeStateOrHostInfoChangeHandler handler = mock(NodeStateOrHostInfoChangeHandler.class);
         NodeInfo nodeInfo = cluster.getNodeInfo(node);
 
-        generator.handleNewReportedNodeState(nodeInfo, nodeState, handler);
+        nodeStateChangeHandler.handleNewReportedNodeState(stateBefore, nodeInfo, nodeState, handler);
         nodeInfo.setReportedState(nodeState, timer.getCurrentTimeInMillis());
     }
 
-    public void reportStorageNodeState(final int index, NodeState nodeState) {
+    ClusterFixture reportStorageNodeState(final int index, State state, String description) {
         final Node node = new Node(NodeType.STORAGE, index);
-        final NodeInfo nodeInfo = cluster.getNodeInfo(node);
-        final long mockTime = 1234;
-        NodeStateOrHostInfoChangeHandler changeListener = mock(NodeStateOrHostInfoChangeHandler.class);
-        generator.handleNewReportedNodeState(nodeInfo, nodeState, changeListener);
-        nodeInfo.setReportedState(nodeState, mockTime);
+        final NodeState nodeState = new NodeState(NodeType.STORAGE, state);
+        nodeState.setDescription(description);
+        doReportNodeState(node, nodeState);
+        return this;
     }
 
-    public void reportDistributorNodeState(final int index, State state) {
+    ClusterFixture reportStorageNodeState(final int index, State state) {
+        return reportStorageNodeState(index, state, "mockdesc");
+    }
+
+    ClusterFixture reportStorageNodeState(final int index, NodeState nodeState) {
+        doReportNodeState(new Node(NodeType.STORAGE, index), nodeState);
+        return this;
+    }
+
+    ClusterFixture reportDistributorNodeState(final int index, State state) {
         final Node node = new Node(NodeType.DISTRIBUTOR, index);
         final NodeState nodeState = new NodeState(NodeType.DISTRIBUTOR, state);
-        NodeStateOrHostInfoChangeHandler handler = mock(NodeStateOrHostInfoChangeHandler.class);
+        doReportNodeState(node, nodeState);
+        return this;
+    }
+
+    ClusterFixture reportDistributorNodeState(final int index, NodeState nodeState) {
+        doReportNodeState(new Node(NodeType.DISTRIBUTOR, index), nodeState);
+        return this;
+    }
+
+    private void doProposeWantedState(final Node node, final NodeState nodeState, String description) {
+        final ClusterState stateBefore = rawGeneratedClusterState();
+
+        nodeState.setDescription(description);
         NodeInfo nodeInfo = cluster.getNodeInfo(node);
+        nodeInfo.setWantedState(nodeState);
 
-        generator.handleNewReportedNodeState(nodeInfo, nodeState, handler);
-        nodeInfo.setReportedState(nodeState, timer.getCurrentTimeInMillis());
+        nodeStateChangeHandler.proposeNewNodeState(stateBefore, nodeInfo, nodeState);
     }
 
-    public void proposeStorageNodeWantedState(final int index, State state) {
+    ClusterFixture proposeStorageNodeWantedState(final int index, State state, String description) {
         final Node node = new Node(NodeType.STORAGE, index);
         final NodeState nodeState = new NodeState(NodeType.STORAGE, state);
+        doProposeWantedState(node, nodeState, description);
+        return this;
+    }
+
+    ClusterFixture proposeStorageNodeWantedState(final int index, State state) {
+        return proposeStorageNodeWantedState(index, state, "mockdesc");
+    }
+
+    ClusterFixture proposeDistributorWantedState(final int index, State state) {
+        final ClusterState stateBefore = rawGeneratedClusterState();
+        final Node node = new Node(NodeType.DISTRIBUTOR, index);
+        final NodeState nodeState = new NodeState(NodeType.DISTRIBUTOR, state);
         nodeState.setDescription("mockdesc");
         NodeInfo nodeInfo = cluster.getNodeInfo(node);
         nodeInfo.setWantedState(nodeState);
 
-        generator.proposeNewNodeState(nodeInfo, nodeState);
+        nodeStateChangeHandler.proposeNewNodeState(stateBefore, nodeInfo, nodeState);
+        return this;
+    }
 
+    ClusterFixture disableAutoClusterTakedown() {
+        setMinNodesUp(0, 0, 0.0, 0.0);
+        return this;
     }
 
-    public void disableAutoClusterTakedown() {
-        generator.setMinNodesUp(0, 0, 0.0, 0.0);
+    ClusterFixture setMinNodesUp(int minDistNodes, int minStorNodes, double minDistRatio, double minStorRatio) {
+        params.minStorageNodesUp(minStorNodes)
+              .minDistributorNodesUp(minDistNodes)
+              .minRatioOfStorageNodesUp(minStorRatio)
+              .minRatioOfDistributorNodesUp(minDistRatio);
+        return this;
     }
 
-    public void disableTransientMaintenanceModeOnDown() {
-        Map<NodeType, Integer> maxTransitionTime = new TreeMap<>();
-        maxTransitionTime.put(NodeType.DISTRIBUTOR, 0);
-        maxTransitionTime.put(NodeType.STORAGE, 0);
-        generator.setMaxTransitionTime(maxTransitionTime);
+    ClusterFixture setMinNodeRatioPerGroup(double upRatio) {
+        params.minNodeRatioPerGroup(upRatio);
+        return this;
     }
 
-    public void enableTransientMaintenanceModeOnDown(final int transitionTime) {
+    static Map<NodeType, Integer> buildTransitionTimeMap(int distributorTransitionTime, int storageTransitionTime) {
         Map<NodeType, Integer> maxTransitionTime = new TreeMap<>();
-        maxTransitionTime.put(NodeType.DISTRIBUTOR, transitionTime);
-        maxTransitionTime.put(NodeType.STORAGE, transitionTime);
-        generator.setMaxTransitionTime(maxTransitionTime);
+        maxTransitionTime.put(NodeType.DISTRIBUTOR, distributorTransitionTime);
+        maxTransitionTime.put(NodeType.STORAGE, storageTransitionTime);
+        return maxTransitionTime;
     }
 
-    public String generatedClusterState() {
-        return generator.getClusterState().toString();
+    void disableTransientMaintenanceModeOnDown() {
+        this.params.transitionTimes(0);
     }
 
-    public String verboseGeneratedClusterState() { return generator.getClusterState().toString(true); }
+    void enableTransientMaintenanceModeOnDown(final int transitionTimeMs) {
+        this.params.transitionTimes(transitionTimeMs);
+    }
+
+    AnnotatedClusterState annotatedGeneratedClusterState() {
+        params.currentTimeInMilllis(timer.getCurrentTimeInMillis());
+        return ClusterStateGenerator.generatedStateFrom(params);
+    }
 
-    public static ClusterFixture forFlatCluster(int nodeCount) {
+    ClusterState rawGeneratedClusterState() {
+        return annotatedGeneratedClusterState().getClusterState();
+    }
+
+    String generatedClusterState() {
+        return annotatedGeneratedClusterState().getClusterState().toString();
+    }
+
+    String verboseGeneratedClusterState() {
+        return annotatedGeneratedClusterState().getClusterState().toString(true);
+    }
+
+    static ClusterFixture forFlatCluster(int nodeCount) {
         Collection<ConfiguredNode> nodes = DistributionBuilder.buildConfiguredNodes(nodeCount);
 
         Distribution distribution = DistributionBuilder.forFlatCluster(nodeCount);
@@ -126,11 +188,27 @@ class ClusterFixture {
         return new ClusterFixture(cluster, distribution);
     }
 
-    public static ClusterFixture forHierarchicCluster(DistributionBuilder.GroupBuilder root) {
+    static ClusterFixture forHierarchicCluster(DistributionBuilder.GroupBuilder root) {
         List<ConfiguredNode> nodes = DistributionBuilder.buildConfiguredNodes(root.totalNodeCount());
         Distribution distribution = DistributionBuilder.forHierarchicCluster(root);
         ContentCluster cluster = new ContentCluster("foo", nodes, distribution, 0, 0.0);
 
         return new ClusterFixture(cluster, distribution);
     }
+
+    ClusterStateGenerator.Params generatorParams() {
+        return new ClusterStateGenerator.Params().cluster(cluster);
+    }
+
+    ContentCluster cluster() {
+        return this.cluster;
+    }
+
+    static Node storageNode(int index) {
+        return new Node(NodeType.STORAGE, index);
+    }
+
+    static Node distributorNode(int index) {
+        return new Node(NodeType.DISTRIBUTOR, index);
+    }
 }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateGeneratorTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateGeneratorTest.java
new file mode 100644
index 00000000000..b9b97c27949
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateGeneratorTest.java
@@ -0,0 +1,895 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+import com.yahoo.vdslib.distribution.ConfiguredNode;
+import com.yahoo.vdslib.state.DiskState;
+import com.yahoo.vdslib.state.Node;
+import com.yahoo.vdslib.state.NodeState;
+import com.yahoo.vdslib.state.NodeType;
+import com.yahoo.vdslib.state.State;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.Optional;
+
+import static com.yahoo.vespa.clustercontroller.core.matchers.HasStateReasonForNode.hasStateReasonForNode;
+import static com.yahoo.vespa.clustercontroller.core.ClusterFixture.storageNode;
+
+import static org.hamcrest.core.IsEqual.equalTo;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+
+public class ClusterStateGeneratorTest {
+
+    private static AnnotatedClusterState generateFromFixtureWithDefaultParams(ClusterFixture fixture) {
+        final ClusterStateGenerator.Params params = new ClusterStateGenerator.Params();
+        params.cluster = fixture.cluster;
+        params.transitionTimes = ClusterFixture.buildTransitionTimeMap(0, 0);
+        params.currentTimeInMillis = 0;
+        return ClusterStateGenerator.generatedStateFrom(params);
+    }
+
+    @Test
+    public void cluster_with_all_nodes_reported_down_has_state_down() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(6).markEntireClusterDown();
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.getClusterState().getClusterState(), is(State.DOWN));
+        // The returned message in this case depends on which "is cluster down?" check
+        // kicks in first. Currently, the minimum storage node count does.
+        assertThat(state.getClusterStateReason(), equalTo(Optional.of(ClusterStateReason.TOO_FEW_STORAGE_NODES_AVAILABLE)));
+    }
+
+    @Test
+    public void cluster_with_all_nodes_up_state_correct_distributor_and_storage_count() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(6).bringEntireClusterUp();
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:6 storage:6"));
+    }
+
+    @Test
+    public void distributor_reported_states_reflected_in_generated_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(9)
+                .bringEntireClusterUp()
+                .reportDistributorNodeState(2, State.DOWN)
+                .reportDistributorNodeState(4, State.STOPPING);
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:9 .2.s:d .4.s:s storage:9"));
+    }
+
+    // NOTE: initializing state tested separately since it involves init progress state info
+    @Test
+    public void storage_reported_states_reflected_in_generated_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(9)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, State.DOWN)
+                .reportStorageNodeState(4, State.STOPPING);
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:9 storage:9 .0.s:d .4.s:s"));
+    }
+
+    @Test
+    public void storage_reported_disk_state_included_in_generated_state() {
+        final NodeState stateWithDisks = new NodeState(NodeType.STORAGE, State.UP);
+        stateWithDisks.setDiskCount(7);
+        stateWithDisks.setDiskState(5, new DiskState(State.DOWN));
+
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(9)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(2, stateWithDisks);
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:9 storage:9 .2.d:7 .2.d.5.s:d"));
+    }
+
+    @Test
+    public void worse_distributor_wanted_state_overrides_reported_state() {
+        // Maintenance mode is illegal for distributors and therefore not tested
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(7)
+                .bringEntireClusterUp()
+                .proposeDistributorWantedState(5, State.DOWN) // Down worse than Up
+                .reportDistributorNodeState(2, State.STOPPING)
+                .proposeDistributorWantedState(2, State.DOWN); // Down worse than Stopping
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:7 .2.s:d .5.s:d storage:7"));
+    }
+
+    @Test
+    public void worse_storage_wanted_state_overrides_reported_state() {
+        // Does not test all maintenance mode overrides; see maintenance_mode_overrides_reported_state
+        // for that.
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(7)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(2, State.STOPPING)
+                .proposeStorageNodeWantedState(2, State.MAINTENANCE) // Maintenance worse than Stopping
+                .proposeStorageNodeWantedState(4, State.RETIRED) // Retired is "worse" than Up
+                .proposeStorageNodeWantedState(5, State.DOWN); // Down worse than Up
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:7 storage:7 .2.s:m .4.s:r .5.s:d"));
+    }
+
+    @Test
+    public void better_distributor_wanted_state_does_not_override_reported_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(7)
+                .bringEntireClusterUp()
+                .reportDistributorNodeState(0, State.DOWN)
+                .proposeDistributorWantedState(0, State.UP);
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:7 .0.s:d storage:7"));
+    }
+
+    @Test
+    public void better_storage_wanted_state_does_not_override_reported_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(7)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(1, State.DOWN)
+                .proposeStorageNodeWantedState(1, State.UP)
+                .reportStorageNodeState(2, State.DOWN)
+                .proposeStorageNodeWantedState(2, State.RETIRED);
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:7 storage:7 .1.s:d .2.s:d"));
+    }
+
+    /**
+     * If we let a Retired node be published as Initializing when it is in init state, we run
+     * the risk of having both feed and merge ops be sent towards it, which is not what we want.
+     * Consequently we pretend such nodes are never in init state and just transition them
+     * directly from Maintenance -> Up.
+     */
+    @Test
+    public void retired_node_in_init_state_is_set_to_maintenance() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(1, State.INITIALIZING)
+                .proposeStorageNodeWantedState(1, State.RETIRED);
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:3 storage:3 .1.s:m"));
+    }
+
+    /**
+     * A storage node will report itself as being in initializing mode immediately when
+     * starting up. It can only accept external operations once it has finished listing
+     * the set of buckets (but not necessarily their contents). As a consequence of this,
+     * we have to map reported init state while bucket listing mode to Down. This will
+     * prevent clients from thinking they can use the node and prevent distributors form
+     * trying to fetch yet non-existent bucket sets from it.
+     *
+     * Detecting the bucket-listing stage is currently done by inspecting its init progress
+     * value and triggering on a sufficiently low value.
+     */
+    @Test
+    public void storage_node_in_init_mode_while_listing_buckets_is_marked_down() {
+        final NodeState initWhileListingBuckets = new NodeState(NodeType.STORAGE, State.INITIALIZING);
+        initWhileListingBuckets.setInitProgress(0.0);
+
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(1, initWhileListingBuckets);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("distributor:3 storage:3 .1.s:d"));
+    }
+
+    /**
+     * Implicit down while reported as init should not kick into effect if the Wanted state
+     * is set to Maintenance.
+     */
+    @Test
+    public void implicit_down_while_listing_buckets_does_not_override_wanted_state() {
+        final NodeState initWhileListingBuckets = new NodeState(NodeType.STORAGE, State.INITIALIZING);
+        initWhileListingBuckets.setInitProgress(0.0);
+
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(1, initWhileListingBuckets)
+                .proposeStorageNodeWantedState(1, State.MAINTENANCE);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("distributor:3 storage:3 .1.s:m"));
+    }
+
+    @Test
+    public void distributor_nodes_in_init_mode_are_not_mapped_to_down() {
+        final NodeState initWhileListingBuckets = new NodeState(NodeType.DISTRIBUTOR, State.INITIALIZING);
+        initWhileListingBuckets.setInitProgress(0.0);
+
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportDistributorNodeState(1, initWhileListingBuckets);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("distributor:3 .1.s:i .1.i:0.0 storage:3"));
+    }
+
+    /**
+     * Maintenance mode overrides all reported states, even Down.
+     */
+    @Test
+    public void maintenance_mode_wanted_state_overrides_reported_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(7)
+                .bringEntireClusterUp()
+                .proposeStorageNodeWantedState(0, State.MAINTENANCE)
+                .reportStorageNodeState(2, State.STOPPING)
+                .proposeStorageNodeWantedState(2, State.MAINTENANCE)
+                .reportStorageNodeState(3, State.DOWN)
+                .proposeStorageNodeWantedState(3, State.MAINTENANCE)
+                .reportStorageNodeState(4, State.INITIALIZING)
+                .proposeStorageNodeWantedState(4, State.MAINTENANCE);
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:7 storage:7 .0.s:m .2.s:m .3.s:m .4.s:m"));
+    }
+
+    @Test
+    public void wanted_state_description_carries_over_to_generated_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(7)
+                .bringEntireClusterUp()
+                .proposeStorageNodeWantedState(1, State.MAINTENANCE, "foo")
+                .proposeStorageNodeWantedState(2, State.DOWN, "bar")
+                .proposeStorageNodeWantedState(3, State.RETIRED, "baz");
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        // We have to use toString(true) to get verbose printing including the descriptions,
+        // as these are omitted by default.
+        assertThat(state.toString(true), equalTo("distributor:7 storage:7 .1.s:m .1.m:foo " +
+                ".2.s:d .2.m:bar .3.s:r .3.m:baz"));
+    }
+
+    @Test
+    public void reported_disk_state_not_hidden_by_wanted_state() {
+        final NodeState stateWithDisks = new NodeState(NodeType.STORAGE, State.UP);
+        stateWithDisks.setDiskCount(5);
+        stateWithDisks.setDiskState(3, new DiskState(State.DOWN));
+
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(9)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(2, stateWithDisks)
+                .proposeStorageNodeWantedState(2, State.RETIRED)
+                .reportStorageNodeState(3, stateWithDisks)
+                .proposeStorageNodeWantedState(3, State.MAINTENANCE);
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        // We do not publish disk states for nodes in Down state. This differs from how the
+        // legacy controller did things, but such states cannot be counted on for ideal state
+        // calculations either way. In particular, reported disk states are not persisted and
+        // only exist transiently in the cluster controller's memory. A controller restart is
+        // sufficient to clear all disk states that have been incidentally remembered for now
+        // downed nodes.
+        // The keen reader may choose to convince themselves of this independently by reading the
+        // code in com.yahoo.vdslib.distribution.Distribution#getIdealStorageNodes and observing
+        // how disk states for nodes that are in a down-state are never considered.
+        assertThat(state.toString(), equalTo("distributor:9 storage:9 .2.s:r .2.d:5 .2.d.3.s:d " +
+                ".3.s:m .3.d:5 .3.d.3.s:d"));
+    }
+
+    @Test
+    public void config_retired_mode_is_reflected_in_generated_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp();
+        List<ConfiguredNode> nodes = DistributionBuilder.buildConfiguredNodes(5);
+        nodes.set(2, new ConfiguredNode(2, true));
+        fixture.cluster.setNodes(nodes);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .2.s:r"));
+    }
+
+    private void do_test_change_within_node_transition_time_window_generates_maintenance(State reportedState) {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp();
+        final ClusterStateGenerator.Params params = fixture.generatorParams()
+                .currentTimeInMilllis(10_000)
+                .transitionTimes(2000);
+
+        fixture.reportStorageNodeState(1, reportedState);
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1));
+        // Node 1 transitioned to reported `reportedState` at time 9000ms after epoch. This means that according to the
+        // above transition time config, it should remain in generated maintenance mode until time 11000ms,
+        // at which point it should finally transition to generated state Down.
+        nodeInfo.setTransitionTime(9000);
+        {
+            final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+            assertThat(state.toString(), equalTo("distributor:5 storage:5 .1.s:m"));
+        }
+
+        nodeInfo.setTransitionTime(10999);
+        {
+            final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+            assertThat(state.toString(), equalTo("distributor:5 storage:5 .1.s:m"));
+        }
+    }
+
+    @Test
+    public void reported_down_node_within_transition_time_has_maintenance_generated_state() {
+        do_test_change_within_node_transition_time_window_generates_maintenance(State.DOWN);
+    }
+
+    @Test
+    public void reported_stopping_node_within_transition_time_has_maintenance_generated_state() {
+        do_test_change_within_node_transition_time_window_generates_maintenance(State.STOPPING);
+    }
+
+    @Test
+    public void reported_node_down_after_transition_time_has_down_generated_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp();
+        final ClusterStateGenerator.Params params = fixture.generatorParams()
+                .currentTimeInMilllis(11_000)
+                .transitionTimes(2000);
+
+        fixture.reportStorageNodeState(1, State.DOWN);
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1));
+        nodeInfo.setTransitionTime(9000);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .1.s:d"));
+    }
+
+    @Test
+    public void distributor_nodes_are_not_implicitly_transitioned_to_maintenance_mode() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp();
+        final ClusterStateGenerator.Params params = fixture.generatorParams()
+                .currentTimeInMilllis(10_000)
+                .transitionTimes(2000);
+
+        fixture.reportDistributorNodeState(2, State.DOWN);
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.DISTRIBUTOR, 2));
+        nodeInfo.setTransitionTime(9000);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:5 .2.s:d storage:5"));
+    }
+
+    @Test
+    public void transient_maintenance_mode_does_not_override_wanted_down_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp();
+        final ClusterStateGenerator.Params params = fixture.generatorParams()
+                .currentTimeInMilllis(10_000)
+                .transitionTimes(2000);
+
+        fixture.proposeStorageNodeWantedState(2, State.DOWN);
+        fixture.reportStorageNodeState(2, State.DOWN);
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 2));
+        nodeInfo.setTransitionTime(9000);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        // Should _not_ be in maintenance mode, since we explicitly want it to stay down.
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .2.s:d"));
+    }
+
+    @Test
+    public void reported_down_retired_node_within_transition_time_transitions_to_maintenance() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp();
+        final ClusterStateGenerator.Params params = fixture.generatorParams()
+                .currentTimeInMilllis(10_000)
+                .transitionTimes(2000);
+
+        fixture.proposeStorageNodeWantedState(2, State.RETIRED);
+        fixture.reportStorageNodeState(2, State.DOWN);
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 2));
+        nodeInfo.setTransitionTime(9000);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .2.s:m"));
+    }
+
+    @Test
+    public void crash_count_exceeding_limit_marks_node_as_down() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp();
+        final ClusterStateGenerator.Params params = fixture.generatorParams().maxPrematureCrashes(10);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 3));
+        nodeInfo.setPrematureCrashCount(11);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .3.s:d"));
+    }
+
+    @Test
+    public void crash_count_not_exceeding_limit_does_not_mark_node_as_down() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp();
+        final ClusterStateGenerator.Params params = fixture.generatorParams().maxPrematureCrashes(10);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 3));
+        nodeInfo.setPrematureCrashCount(10); // "Max crashes" range is inclusive
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5"));
+    }
+
+    @Test
+    public void exceeded_crash_count_does_not_override_wanted_maintenance_state() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5)
+                .bringEntireClusterUp()
+                .proposeStorageNodeWantedState(1, State.MAINTENANCE);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().maxPrematureCrashes(10);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1));
+        nodeInfo.setPrematureCrashCount(11);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .1.s:m"));
+    }
+
+    // Stopping -> Down is expected and does not indicate an unstable node.
+    @Test
+    public void transition_from_controlled_stop_to_down_does_not_add_to_crash_counter() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(2)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(1, State.STOPPING, "controlled shutdown") // urgh, string matching logic
+                .reportStorageNodeState(1, State.DOWN);
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1));
+        assertThat(nodeInfo.getPrematureCrashCount(), equalTo(0));
+    }
+
+    @Test
+    public void non_observed_storage_node_start_timestamp_is_included_in_state() {
+        final NodeState nodeState = new NodeState(NodeType.STORAGE, State.UP);
+        // A reported state timestamp that is not yet marked as observed in the NodeInfo
+        // for the same node is considered not observed by other nodes and must therefore
+        // be included in the generated cluster state
+        nodeState.setStartTimestamp(5000);
+
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, nodeState);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .0.t:5000"));
+    }
+
+    @Test
+    public void non_observed_distributor_start_timestamp_is_included_in_state() {
+        final NodeState nodeState = new NodeState(NodeType.DISTRIBUTOR, State.UP);
+        nodeState.setStartTimestamp(6000);
+
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5)
+                .bringEntireClusterUp()
+                .reportDistributorNodeState(1, nodeState);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("distributor:5 .1.t:6000 storage:5"));
+    }
+
+    @Test
+    public void fully_observed_storage_node_timestamp_not_included_in_state() {
+        final NodeState nodeState = new NodeState(NodeType.STORAGE, State.UP);
+        nodeState.setStartTimestamp(5000);
+
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, nodeState);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 0));
+        nodeInfo.setStartTimestamp(5000);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5"));
+    }
+
+    @Test
+    public void fully_observed_distributor_timestamp_not_included_in_state() {
+        final NodeState nodeState = new NodeState(NodeType.DISTRIBUTOR, State.UP);
+        nodeState.setStartTimestamp(6000);
+
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5)
+                .bringEntireClusterUp()
+                .reportDistributorNodeState(0, nodeState);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.DISTRIBUTOR, 0));
+        nodeInfo.setStartTimestamp(6000);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5"));
+    }
+
+    @Test
+    public void cluster_down_if_less_than_min_count_of_storage_nodes_available() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, State.DOWN)
+                .reportStorageNodeState(2, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minStorageNodesUp(2);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("cluster:d distributor:3 storage:2 .0.s:d"));
+        assertThat(state.getClusterStateReason(), equalTo(Optional.of(ClusterStateReason.TOO_FEW_STORAGE_NODES_AVAILABLE)));
+    }
+
+    @Test
+    public void cluster_not_down_if_more_than_min_count_of_storage_nodes_are_available() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minStorageNodesUp(2);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:3 storage:3 .0.s:d"));
+        assertThat(state.getClusterStateReason(), equalTo(Optional.empty()));
+    }
+
+    @Test
+    public void cluster_down_if_less_than_min_count_of_distributors_available() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportDistributorNodeState(0, State.DOWN)
+                .reportDistributorNodeState(2, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minDistributorNodesUp(2);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("cluster:d distributor:2 .0.s:d storage:3"));
+        assertThat(state.getClusterStateReason(), equalTo(Optional.of(ClusterStateReason.TOO_FEW_DISTRIBUTOR_NODES_AVAILABLE)));
+    }
+
+    @Test
+    public void cluster_not_down_if_more_than_min_count_of_distributors_are_available() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportDistributorNodeState(0, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minDistributorNodesUp(2);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:3 .0.s:d storage:3"));
+        assertThat(state.getClusterStateReason(), equalTo(Optional.empty()));
+    }
+
+    @Test
+    public void maintenance_mode_counted_as_down_for_cluster_availability() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, State.DOWN)
+                .proposeStorageNodeWantedState(2, State.MAINTENANCE);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minStorageNodesUp(2);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("cluster:d distributor:3 storage:3 .0.s:d .2.s:m"));
+    }
+
+    @Test
+    public void init_and_retired_counted_as_up_for_cluster_availability() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, State.INITIALIZING)
+                .proposeStorageNodeWantedState(1, State.RETIRED);
+        // Any node being treated as down should take down the cluster here
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minStorageNodesUp(3);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:3 storage:3 .0.s:i .0.i:1.0 .1.s:r"));
+    }
+
+    @Test
+    public void cluster_down_if_less_than_min_ratio_of_storage_nodes_available() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, State.DOWN)
+                .reportStorageNodeState(2, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minRatioOfStorageNodesUp(0.5);
+
+        // TODO de-dupe a lot of these tests?
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("cluster:d distributor:3 storage:2 .0.s:d"));
+        assertThat(state.getClusterStateReason(), equalTo(Optional.of(ClusterStateReason.TOO_LOW_AVAILABLE_STORAGE_NODE_RATIO)));
+    }
+
+    @Test
+    public void cluster_not_down_if_more_than_min_ratio_of_storage_nodes_available() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, State.DOWN);
+        // Min node ratio is inclusive, i.e. 0.5 of 2 nodes is enough for cluster to be up.
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minRatioOfStorageNodesUp(0.5);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:3 storage:3 .0.s:d"));
+        assertThat(state.getClusterStateReason(), equalTo(Optional.empty()));
+    }
+
+    @Test
+    public void cluster_down_if_less_than_min_ratio_of_distributors_available() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportDistributorNodeState(0, State.DOWN)
+                .reportDistributorNodeState(2, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minRatioOfDistributorNodesUp(0.5);
+
+        // TODO de-dupe a lot of these tests?
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("cluster:d distributor:2 .0.s:d storage:3"));
+        assertThat(state.getClusterStateReason(), equalTo(Optional.of(ClusterStateReason.TOO_LOW_AVAILABLE_DISTRIBUTOR_NODE_RATIO)));
+    }
+
+    @Test
+    public void cluster_not_down_if_more_than_min_ratio_of_distributors_available() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportDistributorNodeState(0, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minRatioOfDistributorNodesUp(0.5);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:3 .0.s:d storage:3"));
+        assertThat(state.getClusterStateReason(), equalTo(Optional.empty()));
+    }
+
+    @Test
+    public void group_nodes_are_marked_down_if_group_availability_too_low() {
+        final ClusterFixture fixture = ClusterFixture
+                .forHierarchicCluster(DistributionBuilder.withGroups(3).eachWithNodeCount(3))
+                .bringEntireClusterUp()
+                .reportStorageNodeState(4, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minNodeRatioPerGroup(0.68);
+
+        // Node 4 is down, which is more than 32% of nodes down in group #2. Nodes 3,5 should be implicitly
+        // marked down as it is in the same group.
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:9 storage:9 .3.s:d .4.s:d .5.s:d"));
+    }
+
+    @Test
+    public void group_nodes_are_not_marked_down_if_group_availability_sufficiently_high() {
+        final ClusterFixture fixture = ClusterFixture
+                .forHierarchicCluster(DistributionBuilder.withGroups(3).eachWithNodeCount(3))
+                .bringEntireClusterUp()
+                .reportStorageNodeState(4, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minNodeRatioPerGroup(0.65);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:9 storage:9 .4.s:d")); // No other nodes down implicitly
+    }
+
+    @Test
+    public void implicitly_downed_group_nodes_receive_a_state_description() {
+        final ClusterFixture fixture = ClusterFixture
+                .forHierarchicCluster(DistributionBuilder.withGroups(2).eachWithNodeCount(2))
+                .bringEntireClusterUp()
+                .reportStorageNodeState(3, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minNodeRatioPerGroup(0.51);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(true), equalTo("distributor:4 storage:4 " +
+                ".2.s:d .2.m:group\\x20node\\x20availability\\x20below\\x20configured\\x20threshold " +
+                ".3.s:d .3.m:mockdesc")); // Preserve description for non-implicitly taken down node
+    }
+
+    @Test
+    public void implicitly_downed_group_nodes_are_annotated_with_group_reason() {
+        final ClusterFixture fixture = ClusterFixture
+                .forHierarchicCluster(DistributionBuilder.withGroups(2).eachWithNodeCount(2))
+                .bringEntireClusterUp()
+                .reportStorageNodeState(3, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minNodeRatioPerGroup(0.51);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.getNodeStateReasons(),
+                hasStateReasonForNode(storageNode(2), NodeStateReason.GROUP_IS_DOWN));
+    }
+
+    @Test
+    public void maintenance_nodes_in_downed_group_are_not_affected() {
+        final ClusterFixture fixture = ClusterFixture
+                .forHierarchicCluster(DistributionBuilder.withGroups(3).eachWithNodeCount(3))
+                .bringEntireClusterUp()
+                .proposeStorageNodeWantedState(3, State.MAINTENANCE)
+                .reportStorageNodeState(4, State.DOWN);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().minNodeRatioPerGroup(0.68);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        // 4 is down by itself, 5 is down implicitly and 3 should happily stay in Maintenance mode.
+        // Side note: most special cases for when a node should and should not be affected by group
+        // down edges are covered in GroupAvailabilityCalculatorTest and GroupAutoTakedownTest.
+        // We test this case explicitly since it's an assurance that code integration works as expected.
+        assertThat(state.toString(), equalTo("distributor:9 storage:9 .3.s:m .4.s:d .5.s:d"));
+    }
+
+    /**
+     * Cluster-wide distribution bit count cannot be higher than the lowest split bit
+     * count reported by the set of storage nodes. This is because the distribution bit
+     * directly impacts which level of the bucket tree is considered the root level,
+     * and any buckets caught over this level would not be accessible in the data space.
+     */
+    @Test
+    public void distribution_bits_bounded_by_reported_min_bits_from_storage_node() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(1, new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(7));
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("bits:7 distributor:3 storage:3"));
+    }
+
+    @Test
+    public void distribution_bits_bounded_by_lowest_reporting_storage_node() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(6))
+                .reportStorageNodeState(1, new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(5));
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("bits:5 distributor:3 storage:3"));
+    }
+
+    @Test
+    public void distribution_bits_bounded_by_config_parameter() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3).bringEntireClusterUp();
+
+        final ClusterStateGenerator.Params params = fixture.generatorParams().idealDistributionBits(12);
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("bits:12 distributor:3 storage:3"));
+    }
+
+    // TODO do we really want this behavior? It's the legacy one, but it seems... dangerous.. Especially for maintenance
+    // TODO We generally want to avoid distribution bit decreases if at all possible, since "collapsing"
+    // the top-level bucket space can cause data loss on timestamp collisions across super buckets.
+    @Test
+    public void distribution_bit_not_influenced_by_nodes_down_or_in_maintenance() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(7))
+                .reportStorageNodeState(1, new NodeState(NodeType.STORAGE, State.DOWN).setMinUsedBits(6))
+                .reportStorageNodeState(2, new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(5))
+                .proposeStorageNodeWantedState(2, State.MAINTENANCE);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("bits:7 distributor:3 storage:3 .1.s:d .2.s:m"));
+    }
+
+    private String do_test_distribution_bit_watermark(int lowestObserved, int node0MinUsedBits) {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(node0MinUsedBits));
+
+        final ClusterStateGenerator.Params params = fixture.generatorParams()
+                .highestObservedDistributionBitCount(8) // TODO is this even needed for our current purposes?
+                .lowestObservedDistributionBitCount(lowestObserved);
+
+        return ClusterStateGenerator.generatedStateFrom(params).toString();
+    }
+
+    /**
+     * Distribution bit increases should not take place incrementally. Doing so would
+     * let e.g. a transition from 10 bits to 20 bits cause 10 interim full re-distributions.
+     */
+    @Test
+    public void published_distribution_bit_bound_by_low_watermark_when_nodes_report_less_than_config_bits() {
+        assertThat(do_test_distribution_bit_watermark(5, 5),
+                   equalTo("bits:5 distributor:3 storage:3"));
+        assertThat(do_test_distribution_bit_watermark(5, 6),
+                   equalTo("bits:5 distributor:3 storage:3"));
+        assertThat(do_test_distribution_bit_watermark(5, 15),
+                   equalTo("bits:5 distributor:3 storage:3"));
+    }
+
+    @Test
+    public void published_state_jumps_to_configured_ideal_bits_when_all_nodes_report_it() {
+        // Note: the rest of the mocked nodes always report 16 bits by default
+        assertThat(do_test_distribution_bit_watermark(5, 16),
+                   equalTo("distributor:3 storage:3")); // "bits:16" implied
+    }
+
+    private String do_test_storage_node_with_no_init_progress(State wantedState) {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, new NodeState(NodeType.STORAGE, State.INITIALIZING).setInitProgress(0.5))
+                .proposeStorageNodeWantedState(0, wantedState);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 0));
+        nodeInfo.setInitProgressTime(10_000);
+
+        final ClusterStateGenerator.Params params = fixture.generatorParams()
+                .maxInitProgressTime(1000)
+                .currentTimeInMilllis(11_000);
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        return state.toString();
+    }
+
+    @Test
+    public void storage_node_with_no_init_progress_within_timeout_is_marked_down() {
+        assertThat(do_test_storage_node_with_no_init_progress(State.UP),
+                   equalTo("distributor:3 storage:3 .0.s:d"));
+    }
+
+    /**
+     * As per usual, we shouldn't transition implicitly to Down if Maintenance is set
+     * as the wanted state.
+     */
+    @Test
+    public void maintenance_wanted_state_overrides_storage_node_with_no_init_progress() {
+        assertThat(do_test_storage_node_with_no_init_progress(State.MAINTENANCE),
+                   equalTo("distributor:3 storage:3 .0.s:m"));
+    }
+
+    /**
+     * Legacy behavior: if a node has crashed (i.e. transition into Down) at least once
+     * while in Init mode, its subsequent init mode will not be made public.
+     * This means the node will remain in a Down-state until it has finished
+     * initializing. This is presumably because unstable nodes may not be able to finish
+     * their init stage and would otherwise pop in and out of the cluster state.
+     */
+    @Test
+    public void unstable_init_storage_node_has_init_state_substituted_by_down() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, State.INITIALIZING)
+                .reportStorageNodeState(0, State.DOWN) // Init -> Down triggers unstable init flag
+                .reportStorageNodeState(0, new NodeState(NodeType.STORAGE, State.INITIALIZING).setInitProgress(0.5));
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .0.s:d"));
+    }
+
+    @Test
+    public void storage_node_with_crashes_but_not_unstable_init_does_not_have_init_state_substituted_by_down() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, new NodeState(NodeType.STORAGE, State.INITIALIZING).setInitProgress(0.5));
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 0));
+        nodeInfo.setPrematureCrashCount(5);
+
+        final AnnotatedClusterState state = generateFromFixtureWithDefaultParams(fixture);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .0.s:i .0.i:0.5"));
+    }
+
+    /**
+     * The generated state must be considered over the Reported state when deciding whether
+     * to override it with the Wanted state. Otherwise, an unstable retired node could have
+     * its generated state be Retired instead of Down. We want it to stay down instead of
+     * potentially contributing additional instability to the cluster.
+     */
+    @Test
+    public void unstable_retired_node_should_be_marked_down() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(5)
+                .bringEntireClusterUp()
+                .proposeStorageNodeWantedState(3, State.RETIRED);
+        final ClusterStateGenerator.Params params = fixture.generatorParams().maxPrematureCrashes(10);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 3));
+        nodeInfo.setPrematureCrashCount(11);
+
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:5 storage:5 .3.s:d"));
+    }
+
+    @Test
+    public void generator_params_can_inherit_values_from_controller_options() {
+        FleetControllerOptions options = new FleetControllerOptions("foocluster");
+        options.maxPrematureCrashes = 1;
+        options.minStorageNodesUp = 2;
+        options.minDistributorNodesUp = 3;
+        options.minRatioOfStorageNodesUp = 0.4;
+        options.minRatioOfDistributorNodesUp = 0.5;
+        options.minNodeRatioPerGroup = 0.6;
+        options.distributionBits = 7;
+        options.maxTransitionTime = ClusterStateGenerator.Params.buildTransitionTimeMap(1000, 2000);
+        final ClusterStateGenerator.Params params = ClusterStateGenerator.Params.fromOptions(options);
+        assertThat(params.maxPrematureCrashes, equalTo(options.maxPrematureCrashes));
+        assertThat(params.minStorageNodesUp, equalTo(options.minStorageNodesUp));
+        assertThat(params.minDistributorNodesUp, equalTo(options.minDistributorNodesUp));
+        assertThat(params.minRatioOfStorageNodesUp, equalTo(options.minRatioOfStorageNodesUp));
+        assertThat(params.minRatioOfDistributorNodesUp, equalTo(options.minRatioOfDistributorNodesUp));
+        assertThat(params.minNodeRatioPerGroup, equalTo(options.minNodeRatioPerGroup));
+        assertThat(params.transitionTimes, equalTo(options.maxTransitionTime));
+    }
+
+    @Test
+    public void configured_zero_init_progress_time_disables_auto_init_to_down_feature() {
+        final ClusterFixture fixture = ClusterFixture.forFlatCluster(3)
+                .bringEntireClusterUp()
+                .reportStorageNodeState(0, new NodeState(NodeType.STORAGE, State.INITIALIZING).setInitProgress(0.5));
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 0));
+        nodeInfo.setInitProgressTime(10_000);
+
+        final ClusterStateGenerator.Params params = fixture.generatorParams()
+                .maxInitProgressTime(0)
+                .currentTimeInMilllis(11_000);
+        final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params);
+        assertThat(state.toString(), equalTo("distributor:3 storage:3 .0.s:i .0.i:0.5"));
+    }
+
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/DistributionBitCountTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/DistributionBitCountTest.java
index 1adb0dcad7d..74661147085 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/DistributionBitCountTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/DistributionBitCountTest.java
@@ -74,13 +74,14 @@ public class DistributionBitCountTest extends FleetControllerTest {
         nodes.get(3).setNodeState(new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(11));
 
         ClusterState startState = waitForState("version:\\d+ bits:11 distributor:10 storage:10");
-        ClusterState state = waitForClusterStateIncludingNodesWithMinUsedBits(11, 2);
 
         nodes.get(1).setNodeState(new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(12));
-        assertEquals(state + "->" + fleetController.getSystemState(), startState.getVersion(), fleetController.getSystemState().getVersion());
+        assertEquals(startState + "->" + fleetController.getSystemState(),
+                     startState.getVersion(), fleetController.getSystemState().getVersion());
 
         for (int i = 0; i < 10; ++i) {
-            nodes.get(i).setNodeState(new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(17));
+            // nodes is array of [distr.0, stor.0, distr.1, stor.1, ...] and we just want the storage nodes
+            nodes.get(i*2 + 1).setNodeState(new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(17));
         }
         assertEquals(startState.getVersion() + 1, waitForState("version:\\d+ bits:17 distributor:10 storage:10").getVersion());
     }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/EventDiffCalculatorTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/EventDiffCalculatorTest.java
new file mode 100644
index 00000000000..2a5b3adcfe7
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/EventDiffCalculatorTest.java
@@ -0,0 +1,319 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+import static com.yahoo.vespa.clustercontroller.core.matchers.EventForNode.eventForNode;
+import static com.yahoo.vespa.clustercontroller.core.matchers.NodeEventWithDescription.nodeEventWithDescription;
+import static com.yahoo.vespa.clustercontroller.core.matchers.ClusterEventWithDescription.clusterEventWithDescription;
+import static com.yahoo.vespa.clustercontroller.core.matchers.EventTypeIs.eventTypeIs;
+import static com.yahoo.vespa.clustercontroller.core.matchers.EventTimeIs.eventTimeIs;
+import static org.hamcrest.CoreMatchers.allOf;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+import static org.hamcrest.CoreMatchers.hasItem;
+
+import static com.yahoo.vespa.clustercontroller.core.ClusterFixture.storageNode;
+import static com.yahoo.vespa.clustercontroller.core.ClusterFixture.distributorNode;
+
+import com.yahoo.vdslib.state.ClusterState;
+import com.yahoo.vdslib.state.Node;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public class EventDiffCalculatorTest {
+
+    private static Map<Node, NodeStateReason> emptyNodeStateReasons() {
+        return Collections.emptyMap();
+    }
+
+    private static class EventFixture {
+        final ClusterFixture clusterFixture;
+        // TODO could reasonably put shared state into a common class to avoid dupes for both before/after
+        Optional<ClusterStateReason> clusterReasonBefore = Optional.empty();
+        Optional<ClusterStateReason> clusterReasonAfter = Optional.empty();
+        ClusterState clusterStateBefore = ClusterState.emptyState();
+        ClusterState clusterStateAfter = ClusterState.emptyState();
+        final Map<Node, NodeStateReason> nodeReasonsBefore = new HashMap<>();
+        final Map<Node, NodeStateReason> nodeReasonsAfter = new HashMap<>();
+        long currentTimeMs = 0;
+
+        EventFixture(int nodeCount) {
+            this.clusterFixture = ClusterFixture.forFlatCluster(nodeCount);
+        }
+
+        EventFixture clusterStateBefore(String stateStr) {
+            clusterStateBefore = ClusterState.stateFromString(stateStr);
+            return this;
+        }
+        EventFixture clusterStateAfter(String stateStr) {
+            clusterStateAfter = ClusterState.stateFromString(stateStr);
+            return this;
+        }
+        EventFixture storageNodeReasonBefore(int index, NodeStateReason reason) {
+            nodeReasonsBefore.put(storageNode(index), reason);
+            return this;
+        }
+        EventFixture storageNodeReasonAfter(int index, NodeStateReason reason) {
+            nodeReasonsAfter.put(storageNode(index), reason);
+            return this;
+        }
+        EventFixture clusterReasonBefore(ClusterStateReason reason) {
+            this.clusterReasonBefore = Optional.of(reason);
+            return this;
+        }
+        EventFixture clusterReasonAfter(ClusterStateReason reason) {
+            this.clusterReasonAfter = Optional.of(reason);
+            return this;
+        }
+        EventFixture currentTimeMs(long timeMs) {
+            this.currentTimeMs = timeMs;
+            return this;
+        }
+
+        List<Event> computeEventDiff() {
+            final AnnotatedClusterState stateBefore = new AnnotatedClusterState(
+                    clusterStateBefore, clusterReasonBefore, nodeReasonsBefore);
+            final AnnotatedClusterState stateAfter = new AnnotatedClusterState(
+                    clusterStateAfter, clusterReasonAfter, nodeReasonsAfter);
+
+            return EventDiffCalculator.computeEventDiff(
+                    EventDiffCalculator.params()
+                            .cluster(clusterFixture.cluster())
+                            .fromState(stateBefore)
+                            .toState(stateAfter)
+                            .currentTimeMs(currentTimeMs));
+        }
+
+        static EventFixture createForNodes(int nodeCount) {
+            return new EventFixture(nodeCount);
+        }
+
+    }
+
+    @Test
+    public void single_storage_node_state_transition_emits_altered_node_state_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("distributor:3 storage:3 .0.s:d");
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        assertThat(events, hasItem(allOf(
+                eventForNode(storageNode(0)),
+                eventTypeIs(NodeEvent.Type.CURRENT),
+                nodeEventWithDescription("Altered node state in cluster state from 'U' to 'D'"))));
+    }
+
+    @Test
+    public void single_distributor_node_state_transition_emits_altered_node_state_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("distributor:3 .1.s:d storage:3");
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        assertThat(events, hasItem(allOf(
+                eventForNode(distributorNode(1)),
+                eventTypeIs(NodeEvent.Type.CURRENT),
+                nodeEventWithDescription("Altered node state in cluster state from 'U' to 'D'"))));
+    }
+
+    @Test
+    public void node_state_change_event_is_tagged_with_given_time() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("distributor:3 storage:3 .0.s:d")
+                .currentTimeMs(123456);
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        assertThat(events, hasItem(eventTimeIs(123456)));
+    }
+
+    @Test
+    public void multiple_node_state_transitions_emit_multiple_node_state_events() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3 .1.s:d")
+                .clusterStateAfter("distributor:3 .2.s:d storage:3 .0.s:r");
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(3));
+        assertThat(events, hasItem(allOf(
+                eventForNode(distributorNode(2)),
+                nodeEventWithDescription("Altered node state in cluster state from 'U' to 'D'"))));
+        assertThat(events, hasItem(allOf(
+                eventForNode(storageNode(0)),
+                nodeEventWithDescription("Altered node state in cluster state from 'U' to 'R'"))));
+        assertThat(events, hasItem(allOf(
+                eventForNode(storageNode(1)),
+                nodeEventWithDescription("Altered node state in cluster state from 'D' to 'U'"))));
+    }
+
+    @Test
+    public void no_emitted_node_state_event_when_node_state_not_changed() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("distributor:3 storage:3");
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(0));
+    }
+
+    @Test
+    public void node_down_edge_with_group_down_reason_has_separate_event_emitted() {
+        // We sneakily use a flat cluster here but still use a 'group down' reason. Differ doesn't currently care.
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("distributor:3 storage:3 .1.s:d")
+                .storageNodeReasonAfter(1, NodeStateReason.GROUP_IS_DOWN);
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(2));
+        // Both the regular edge event and the group down event is emitted
+        assertThat(events, hasItem(allOf(
+                eventForNode(storageNode(1)),
+                nodeEventWithDescription("Altered node state in cluster state from 'U' to 'D'"))));
+        assertThat(events, hasItem(allOf(
+                eventForNode(storageNode(1)),
+                eventTypeIs(NodeEvent.Type.CURRENT),
+                nodeEventWithDescription("Group node availability is below configured threshold"))));
+    }
+
+    @Test
+    public void group_down_to_group_down_does_not_emit_new_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3 .1.s:d")
+                .clusterStateAfter("distributor:3 storage:3 .1.s:m")
+                .storageNodeReasonBefore(1, NodeStateReason.GROUP_IS_DOWN)
+                .storageNodeReasonAfter(1, NodeStateReason.GROUP_IS_DOWN);
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        // Should not get a group availability event since nothing has changed in this regard
+        assertThat(events, hasItem(allOf(
+                eventForNode(storageNode(1)),
+                nodeEventWithDescription("Altered node state in cluster state from 'D' to 'M'"))));
+    }
+
+    @Test
+    public void group_down_to_clear_reason_emits_group_up_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3 .2.s:d")
+                .clusterStateAfter("distributor:3 storage:3")
+                .storageNodeReasonBefore(2, NodeStateReason.GROUP_IS_DOWN); // But no after-reason.
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(2));
+        assertThat(events, hasItem(allOf(
+                eventForNode(storageNode(2)),
+                nodeEventWithDescription("Altered node state in cluster state from 'D' to 'U'"))));
+        assertThat(events, hasItem(allOf(
+                eventForNode(storageNode(2)),
+                eventTypeIs(NodeEvent.Type.CURRENT),
+                nodeEventWithDescription("Group node availability has been restored"))));
+    }
+
+    @Test
+    public void cluster_up_edge_emits_sufficient_node_availability_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("cluster:d distributor:3 storage:3")
+                .clusterStateAfter("distributor:3 storage:3");
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        assertThat(events, hasItem(
+                clusterEventWithDescription("Enough nodes available for system to become up")));
+    }
+
+    @Test
+    public void cluster_down_event_without_reason_annotation_emits_generic_down_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("cluster:d distributor:3 storage:3");
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        assertThat(events, hasItem(
+                clusterEventWithDescription("Cluster is down")));
+    }
+
+    @Test
+    public void cluster_event_is_tagged_with_given_time() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("cluster:d distributor:3 storage:3")
+                .currentTimeMs(56789);
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        assertThat(events, hasItem(eventTimeIs(56789)));
+    }
+
+    @Test
+    public void no_event_emitted_for_cluster_down_to_down_edge() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("cluster:d distributor:3 storage:3")
+                .clusterStateAfter("cluster:d distributor:3 storage:3");
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(0));
+    }
+
+    @Test
+    public void too_few_storage_nodes_cluster_down_reason_emits_corresponding_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("cluster:d distributor:3 storage:3")
+                .clusterReasonAfter(ClusterStateReason.TOO_FEW_STORAGE_NODES_AVAILABLE);
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        // TODO(?) these messages currently don't include the current configured limits
+        assertThat(events, hasItem(
+                clusterEventWithDescription("Too few storage nodes available in cluster. Setting cluster state down")));
+    }
+
+    @Test
+    public void too_few_distributor_nodes_cluster_down_reason_emits_corresponding_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("cluster:d distributor:3 storage:3")
+                .clusterReasonAfter(ClusterStateReason.TOO_FEW_DISTRIBUTOR_NODES_AVAILABLE);
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        assertThat(events, hasItem(
+                clusterEventWithDescription("Too few distributor nodes available in cluster. Setting cluster state down")));
+    }
+
+    @Test
+    public void too_low_storage_node_ratio_cluster_down_reason_emits_corresponding_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("cluster:d distributor:3 storage:3")
+                .clusterReasonAfter(ClusterStateReason.TOO_LOW_AVAILABLE_STORAGE_NODE_RATIO);
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        assertThat(events, hasItem(
+                clusterEventWithDescription("Too low ratio of available storage nodes. Setting cluster state down")));
+    }
+
+    @Test
+    public void too_low_distributor_node_ratio_cluster_down_reason_emits_corresponding_event() {
+        final EventFixture fixture = EventFixture.createForNodes(3)
+                .clusterStateBefore("distributor:3 storage:3")
+                .clusterStateAfter("cluster:d distributor:3 storage:3")
+                .clusterReasonAfter(ClusterStateReason.TOO_LOW_AVAILABLE_DISTRIBUTOR_NODE_RATIO);
+
+        final List<Event> events = fixture.computeEventDiff();
+        assertThat(events.size(), equalTo(1));
+        assertThat(events, hasItem(
+                clusterEventWithDescription("Too low ratio of available distributor nodes. Setting cluster state down")));
+    }
+
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/FleetControllerTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/FleetControllerTest.java
index f4b3e648f63..d0aa0bceba9 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/FleetControllerTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/FleetControllerTest.java
@@ -6,13 +6,11 @@ import com.yahoo.jrt.slobrok.server.Slobrok;
 import com.yahoo.log.LogLevel;
 import com.yahoo.log.LogSetup;
 import com.yahoo.vdslib.distribution.ConfiguredNode;
-import com.yahoo.vdslib.distribution.Distribution;
 import com.yahoo.vdslib.state.ClusterState;
 import com.yahoo.vdslib.state.Node;
 import com.yahoo.vdslib.state.NodeState;
 import com.yahoo.vdslib.state.NodeType;
 import com.yahoo.vespa.clustercontroller.core.database.DatabaseHandler;
-import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo;
 import com.yahoo.vespa.clustercontroller.core.rpc.RPCCommunicator;
 import com.yahoo.vespa.clustercontroller.core.rpc.RpcServer;
 import com.yahoo.vespa.clustercontroller.core.rpc.SlobrokClient;
@@ -150,7 +148,7 @@ public abstract class FleetControllerTest implements Waiter {
         }
         RpcServer rpcServer = new RpcServer(timer, timer, options.clusterName, options.fleetControllerIndex, options.slobrokBackOffPolicy);
         DatabaseHandler database = new DatabaseHandler(timer, options.zooKeeperServerAddress, options.fleetControllerIndex, timer);
-        SystemStateGenerator stateGenerator = new SystemStateGenerator(timer, log, metricUpdater);
+        StateChangeHandler stateGenerator = new StateChangeHandler(timer, log, metricUpdater);
         SystemStateBroadcaster stateBroadcaster = new SystemStateBroadcaster(timer, timer);
         MasterElectionHandler masterElectionHandler = new MasterElectionHandler(options.fleetControllerIndex, options.fleetControllerCount, timer, timer);
         FleetController controller = new FleetController(timer, log, cluster, stateGatherer, communicator, status, rpcServer, lookUp, database, stateGenerator, stateBroadcaster, masterElectionHandler, metricUpdater, options);
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/GroupAutoTakedownTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/GroupAutoTakedownTest.java
index be60fba234a..a7307e0180a 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/GroupAutoTakedownTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/GroupAutoTakedownTest.java
@@ -9,19 +9,22 @@ import com.yahoo.vdslib.state.NodeType;
 import com.yahoo.vdslib.state.State;
 import com.yahoo.vespa.clustercontroller.core.database.DatabaseHandler;
 import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler;
-import com.yahoo.vespa.clustercontroller.core.listeners.SystemStateListener;
+
+import static com.yahoo.vespa.clustercontroller.core.matchers.EventForNode.eventForNode;
+import static com.yahoo.vespa.clustercontroller.core.matchers.NodeEventWithDescription.nodeEventWithDescription;
 import org.junit.Test;
-import org.mockito.ArgumentMatcher;
 
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 import static org.hamcrest.core.AllOf.allOf;
+import static org.hamcrest.core.IsCollectionContaining.hasItem;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.argThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -43,26 +46,29 @@ public class GroupAutoTakedownTest {
     }
 
     private static void setSharedFixtureOptions(ClusterFixture fixture, double minNodeRatioPerGroup) {
-        fixture.generator.setMinNodeRatioPerGroup(minNodeRatioPerGroup);
+        fixture.setMinNodeRatioPerGroup(minNodeRatioPerGroup);
         fixture.disableTransientMaintenanceModeOnDown();
         fixture.disableAutoClusterTakedown();
         fixture.bringEntireClusterUp();
     }
 
     private String stateAfterStorageTransition(ClusterFixture fixture, final int index, final State state) {
-        transitionStoreNodeToState(fixture, index, state);
+        transitionStorageNodeToState(fixture, index, state);
         return fixture.generatedClusterState();
     }
 
     private String verboseStateAfterStorageTransition(ClusterFixture fixture, final int index, final State state) {
-        transitionStoreNodeToState(fixture, index, state);
+        transitionStorageNodeToState(fixture, index, state);
         return fixture.verboseGeneratedClusterState();
     }
 
-    private void transitionStoreNodeToState(ClusterFixture fixture, int index, State state) {
+    private void transitionStorageNodeToState(ClusterFixture fixture, int index, State state) {
         fixture.reportStorageNodeState(index, state);
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
+    }
+
+    private AnnotatedClusterState annotatedStateAfterStorageTransition(ClusterFixture fixture, final int index, final State state) {
+        transitionStorageNodeToState(fixture, index, state);
+        return fixture.annotatedGeneratedClusterState();
     }
 
     /**
@@ -74,12 +80,9 @@ public class GroupAutoTakedownTest {
     public void config_does_not_apply_to_flat_hierarchy_clusters() {
         ClusterFixture fixture = createFixtureForAllUpFlatCluster(5, 0.99);
 
-        SystemStateListener listener = mock(SystemStateListener.class);
-        // First invocation; generates initial state and clears "new state" flag
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-        assertEquals("version:1 distributor:5 storage:5", fixture.generatedClusterState());
+        assertEquals("distributor:5 storage:5", fixture.generatedClusterState());
 
-        assertEquals("version:2 distributor:5 storage:5 .1.s:d",
+        assertEquals("distributor:5 storage:5 .1.s:d",
                      stateAfterStorageTransition(fixture, 1, State.DOWN));
     }
 
@@ -88,15 +91,13 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-        assertEquals("version:1 distributor:6 storage:6", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6", fixture.generatedClusterState());
 
         // Same group as node 4
-        assertEquals("version:2 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                      stateAfterStorageTransition(fixture, 5, State.DOWN));
         // Same group as node 1
-        assertEquals("version:3 distributor:6 storage:4 .0.s:d .1.s:d",
+        assertEquals("distributor:6 storage:4 .0.s:d .1.s:d",
                      stateAfterStorageTransition(fixture, 0, State.DOWN));
     }
 
@@ -106,11 +107,11 @@ public class GroupAutoTakedownTest {
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
         // Group #2 -> down
-        assertEquals("version:1 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                      stateAfterStorageTransition(fixture, 5, State.DOWN));
 
         // Group #2 -> back up again
-        assertEquals("version:2 distributor:6 storage:6",
+        assertEquals("distributor:6 storage:6",
                      stateAfterStorageTransition(fixture, 5, State.UP));
     }
 
@@ -119,16 +120,12 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        assertEquals("version:1 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
 
         // 4, 5 in same group; this should not cause a new state since it's already implicitly down
         fixture.reportStorageNodeState(4, State.DOWN);
-
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertFalse(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-
-        assertEquals("version:1 distributor:6 storage:4", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:4", fixture.generatedClusterState());
     }
 
     @Test
@@ -139,7 +136,7 @@ public class GroupAutoTakedownTest {
         // Nodes 6 and 7 are taken down implicitly and should have a message reflecting this.
         // Node 8 is taken down by the fixture and gets a fixture-assigned message that
         // we should _not_ lose/overwrite.
-        assertEquals("version:1 distributor:9 storage:9 .6.s:d " +
+        assertEquals("distributor:9 storage:9 .6.s:d " +
                         ".6.m:group\\x20node\\x20availability\\x20below\\x20configured\\x20threshold " +
                         ".7.s:d " +
                         ".7.m:group\\x20node\\x20availability\\x20below\\x20configured\\x20threshold " +
@@ -151,12 +148,12 @@ public class GroupAutoTakedownTest {
     public void legacy_cluster_wide_availabilty_ratio_is_computed_after_group_takedowns() {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
-        fixture.generator.setMinNodesUp(5, 5, 0.51, 0.51);
+        fixture.setMinNodesUp(5, 5, 0.51, 0.51);
 
         // Taking down a node in a group forces the entire group down, which leaves us with
         // only 4 content nodes (vs. minimum of 5 as specified above). The entire cluster
         // should be marked as down in this case.
-        assertEquals("version:1 cluster:d distributor:6 storage:4",
+        assertEquals("cluster:d distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
     }
 
@@ -165,16 +162,12 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(3), 0.99);
 
-        NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 5));
-        fixture.generator.proposeNewNodeState(nodeInfo, new NodeState(NodeType.STORAGE, State.MAINTENANCE));
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-
+        fixture.proposeStorageNodeWantedState(5, State.MAINTENANCE);
         // Maintenance not counted as down, so group still up
-        assertEquals("version:1 distributor:9 storage:9 .5.s:m", fixture.generatedClusterState());
+        assertEquals("distributor:9 storage:9 .5.s:m", fixture.generatedClusterState());
 
         // Group goes down, but maintenance node should still be in maintenance
-        assertEquals("version:2 distributor:9 storage:9 .3.s:d .4.s:d .5.s:m",
+        assertEquals("distributor:9 storage:9 .3.s:d .4.s:d .5.s:m",
                 stateAfterStorageTransition(fixture, 4, State.DOWN));
     }
 
@@ -186,51 +179,16 @@ public class GroupAutoTakedownTest {
 
         // Our timers are mocked, so taking down node 4 will deterministically transition to
         // a transient maintenance mode. Group should not be taken down here.
-        assertEquals("version:1 distributor:9 storage:9 .4.s:m",
+        assertEquals("distributor:9 storage:9 .4.s:m",
                 stateAfterStorageTransition(fixture, 4, State.DOWN));
 
         // However, once grace period expires the group should be taken down.
         fixture.timer.advanceTime(1001);
         NodeStateOrHostInfoChangeHandler changeListener = mock(NodeStateOrHostInfoChangeHandler.class);
-        fixture.generator.watchTimers(fixture.cluster, changeListener);
-        SystemStateListener stateListener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, stateListener));
-
-        assertEquals("version:2 distributor:9 storage:9 .3.s:d .4.s:d .5.s:d", fixture.generatedClusterState());
-    }
-
-    private static class NodeEventWithDescription extends ArgumentMatcher<NodeEvent> {
-        private final String expected;
-
-        NodeEventWithDescription(String expected) {
-            this.expected = expected;
-        }
-
-        @Override
-        public boolean matches(Object o) {
-            return expected.equals(((NodeEvent)o).getDescription());
-        }
-    }
+        fixture.nodeStateChangeHandler.watchTimers(
+                fixture.cluster, fixture.annotatedGeneratedClusterState().getClusterState(), changeListener);
 
-    private static NodeEventWithDescription nodeEventWithDescription(String description) {
-        return new NodeEventWithDescription(description);
-    }
-
-    private static class EventForNode extends ArgumentMatcher<NodeEvent> {
-        private final Node expected;
-
-        EventForNode(Node expected) {
-            this.expected = expected;
-        }
-
-        @Override
-        public boolean matches(Object o) {
-            return ((NodeEvent)o).getNode().getNode().equals(expected);
-        }
-    }
-
-    private static EventForNode eventForNode(Node expected) {
-        return new EventForNode(expected);
+        assertEquals("distributor:9 storage:9 .3.s:d .4.s:d .5.s:d", fixture.generatedClusterState());
     }
 
     private static Node contentNode(int index) {
@@ -242,13 +200,14 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        assertEquals("version:1 distributor:6 storage:4",
-                stateAfterStorageTransition(fixture, 5, State.DOWN));
+        final List<Event> events = EventDiffCalculator.computeEventDiff(EventDiffCalculator.params()
+                .cluster(fixture.cluster)
+                .fromState(fixture.annotatedGeneratedClusterState())
+                .toState(annotatedStateAfterStorageTransition(fixture, 5, State.DOWN)));
 
-        verify(fixture.eventLog).addNodeOnlyEvent(argThat(allOf(
-                nodeEventWithDescription("Setting node down as the total availability of its group is " +
-                        "below the configured threshold"),
-                eventForNode(contentNode(4)))), any());
+        assertThat(events, hasItem(allOf(
+                nodeEventWithDescription("Group node availability is below configured threshold"),
+                eventForNode(contentNode(4)))));
     }
 
     @Test
@@ -256,30 +215,31 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        assertEquals("version:1 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
-        assertEquals("version:2 distributor:6 storage:6",
-                stateAfterStorageTransition(fixture, 5, State.UP));
 
-        verify(fixture.eventLog).addNodeOnlyEvent(argThat(allOf(
-                nodeEventWithDescription("Group availability restored; taking node back up"),
-                eventForNode(contentNode(4)))), any());
+        final List<Event> events = EventDiffCalculator.computeEventDiff(EventDiffCalculator.params()
+                .cluster(fixture.cluster)
+                .fromState(fixture.annotatedGeneratedClusterState())
+                .toState(annotatedStateAfterStorageTransition(fixture, 5, State.UP)));
+
+        assertThat(events, hasItem(allOf(
+                nodeEventWithDescription("Group node availability has been restored"),
+                eventForNode(contentNode(4)))));
     }
 
     @Test
-    public void wanted_state_retired_implicitly_down_node_transitioned_it_to_retired_mode_immediately() {
+    public void wanted_state_retired_implicitly_down_node_is_transitioned_to_retired_mode_immediately() {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(3), 0.99);
 
-        assertEquals("version:1 distributor:9 storage:6",
+        assertEquals("distributor:9 storage:6",
                 stateAfterStorageTransition(fixture, 6, State.DOWN));
         // Node 7 is implicitly down. Mark wanted state as retired. It should now be Retired
         // but not Down.
         fixture.proposeStorageNodeWantedState(7, State.RETIRED);
 
-        SystemStateListener stateListener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, stateListener));
-        assertEquals("version:2 distributor:9 storage:8 .6.s:d .7.s:r", fixture.generatedClusterState());
+        assertEquals("distributor:9 storage:8 .6.s:d .7.s:r", fixture.generatedClusterState());
     }
 
     @Test
@@ -287,9 +247,9 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.49);
 
-        assertEquals("version:1 distributor:6 storage:6 .4.s:d",
+        assertEquals("distributor:6 storage:6 .4.s:d",
                 stateAfterStorageTransition(fixture, 4, State.DOWN));
-        assertEquals("version:2 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
 
         // Node 5 gets config-retired under our feet.
@@ -299,9 +259,8 @@ public class GroupAutoTakedownTest {
         // TODO this should ideally also set the retired flag in the distribution
         // config, but only the ConfiguredNodes are actually looked at currently.
         fixture.cluster.setNodes(nodes);
-        fixture.generator.setNodes(fixture.cluster.clusterInfo());
 
-        assertEquals("version:3 distributor:6 storage:6 .4.s:d .5.s:r",
+        assertEquals("distributor:6 storage:6 .4.s:d .5.s:r",
                 stateAfterStorageTransition(fixture, 5, State.UP));
     }
 
@@ -314,14 +273,12 @@ public class GroupAutoTakedownTest {
         newState.setInitProgress(0.5);
 
         fixture.reportStorageNodeState(4, newState);
-        SystemStateListener stateListener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, stateListener));
 
-        assertEquals("version:1 distributor:6 storage:6 .4.s:i .4.i:0.5", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6 .4.s:i .4.i:0.5", fixture.generatedClusterState());
 
-        assertEquals("version:2 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
-        assertEquals("version:3 distributor:6 storage:6 .4.s:i .4.i:0.5",
+        assertEquals("distributor:6 storage:6 .4.s:i .4.i:0.5",
                 stateAfterStorageTransition(fixture, 5, State.UP));
     }
 
@@ -330,20 +287,17 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        final Node node = new Node(NodeType.STORAGE, 4);
         final NodeState newState = new NodeState(NodeType.STORAGE, State.UP);
         newState.setDiskCount(7);
         newState.setDiskState(5, new DiskState(State.DOWN));
 
         fixture.reportStorageNodeState(4, newState);
-        SystemStateListener stateListener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, stateListener));
 
-        assertEquals("version:1 distributor:6 storage:6 .4.d:7 .4.d.5.s:d", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6 .4.d:7 .4.d.5.s:d", fixture.generatedClusterState());
 
-        assertEquals("version:2 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
-        assertEquals("version:3 distributor:6 storage:6 .4.d:7 .4.d.5.s:d",
+        assertEquals("distributor:6 storage:6 .4.d:7 .4.d.5.s:d",
                 stateAfterStorageTransition(fixture, 5, State.UP));
     }
 
@@ -352,19 +306,15 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(3), 0.60);
 
-        NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 5));
-        nodeInfo.setWantedState(new NodeState(NodeType.STORAGE, State.DOWN).setDescription("borkbork"));
-        fixture.generator.proposeNewNodeState(nodeInfo, nodeInfo.getWantedState());
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
+        fixture.proposeStorageNodeWantedState(5, State.DOWN, "borkbork");
 
-        assertEquals("version:1 distributor:9 storage:9 .5.s:d .5.m:borkbork", fixture.verboseGeneratedClusterState());
+        assertEquals("distributor:9 storage:9 .5.s:d .5.m:borkbork", fixture.verboseGeneratedClusterState());
 
-        assertEquals("version:2 distributor:9 storage:9 " +
+        assertEquals("distributor:9 storage:9 " +
                 ".3.s:d .3.m:group\\x20node\\x20availability\\x20below\\x20configured\\x20threshold " +
                 ".4.s:d .4.m:mockdesc .5.s:d .5.m:borkbork",
                 verboseStateAfterStorageTransition(fixture, 4, State.DOWN));
-        assertEquals("version:3 distributor:9 storage:9 .5.s:d .5.m:borkbork",
+        assertEquals("distributor:9 storage:9 .5.s:d .5.m:borkbork",
                 verboseStateAfterStorageTransition(fixture, 4, State.UP));
     }
 
@@ -378,25 +328,23 @@ public class GroupAutoTakedownTest {
 
         fixture.reportStorageNodeState(4, newState);
 
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-
-        assertEquals("version:1 distributor:6 storage:6 .4.t:123456", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6 .4.t:123456", fixture.generatedClusterState());
 
         DatabaseHandler handler = mock(DatabaseHandler.class);
         DatabaseHandler.Context context = mock(DatabaseHandler.Context.class);
         when(context.getCluster()).thenReturn(fixture.cluster);
 
-        fixture.generator.handleAllDistributorsInSync(handler, context);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
+        Set<ConfiguredNode> nodes = new HashSet<>(fixture.cluster.clusterInfo().getConfiguredNodes().values());
+        fixture.nodeStateChangeHandler.handleAllDistributorsInSync(
+                fixture.annotatedGeneratedClusterState().getClusterState(), nodes, handler, context);
 
         // Timestamp should now be cleared from state
-        assertEquals("version:2 distributor:6 storage:6", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6", fixture.generatedClusterState());
 
         // Trigger a group down+up edge. Timestamp should _not_ be reintroduced since it was previously cleared.
-        assertEquals("version:3 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
-        assertEquals("version:4 distributor:6 storage:6",
+        assertEquals("distributor:6 storage:6",
                 stateAfterStorageTransition(fixture, 5, State.UP));
     }
 
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/MasterElectionTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/MasterElectionTest.java
index ba2cd287a9a..80435ee7c7d 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/MasterElectionTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/MasterElectionTest.java
@@ -191,15 +191,15 @@ public class MasterElectionTest extends FleetControllerTest {
         log.log(LogLevel.INFO, "Leaving waitForMaster");
     }
 
-    private static class VersionMonotonicityChecker {
+    private static class StrictlyIncreasingVersionChecker {
         private ClusterState lastState;
 
-        private VersionMonotonicityChecker(ClusterState initialState) {
+        private StrictlyIncreasingVersionChecker(ClusterState initialState) {
             this.lastState = initialState;
         }
 
-        public static VersionMonotonicityChecker bootstrappedWith(ClusterState initialState) {
-            return new VersionMonotonicityChecker(initialState);
+        public static StrictlyIncreasingVersionChecker bootstrappedWith(ClusterState initialState) {
+            return new StrictlyIncreasingVersionChecker(initialState);
         }
 
         public void updateAndVerify(ClusterState currentState) {
@@ -207,7 +207,7 @@ public class MasterElectionTest extends FleetControllerTest {
             lastState = currentState;
             if (currentState.getVersion() <= last.getVersion()) {
                 throw new IllegalStateException(
-                        String.format("Cluster state version monotonicity invariant broken! " +
+                        String.format("Cluster state version strict increase invariant broken! " +
                                       "Old state was '%s', new state is '%s'", last, currentState));
             }
         }
@@ -226,7 +226,8 @@ public class MasterElectionTest extends FleetControllerTest {
         waitForStableSystem();
         waitForMaster(0);
         Arrays.asList(0, 1, 2, 3, 4).stream().forEach(this::waitForCompleteCycle);
-        VersionMonotonicityChecker checker = VersionMonotonicityChecker.bootstrappedWith(fleetControllers.get(0).getClusterState());
+        StrictlyIncreasingVersionChecker checker = StrictlyIncreasingVersionChecker.bootstrappedWith(
+                fleetControllers.get(0).getClusterState());
         fleetControllers.get(0).shutdown();
         waitForMaster(1);
         Arrays.asList(1, 2, 3, 4).stream().forEach(this::waitForCompleteCycle);
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/NodeInfoTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/NodeInfoTest.java
new file mode 100644
index 00000000000..bf0adf7736c
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/NodeInfoTest.java
@@ -0,0 +1,80 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+import com.yahoo.vdslib.state.Node;
+import com.yahoo.vdslib.state.NodeType;
+import com.yahoo.vdslib.state.State;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class NodeInfoTest {
+
+    @Test
+    public void unstable_init_flag_is_initially_clear() {
+        ClusterFixture fixture = ClusterFixture.forFlatCluster(3);
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1));
+        assertFalse(nodeInfo.recentlyObservedUnstableDuringInit());
+    }
+
+    private static ClusterFixture fixtureWithNodeMarkedAsUnstableInit(int nodeIndex) {
+        return ClusterFixture.forFlatCluster(3)
+                .reportStorageNodeState(nodeIndex, State.INITIALIZING)
+                .reportStorageNodeState(nodeIndex, State.DOWN);
+    }
+
+    @Test
+    public void down_edge_during_init_state_marks_as_unstable_init() {
+        ClusterFixture fixture = fixtureWithNodeMarkedAsUnstableInit(1);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1));
+        assertTrue(nodeInfo.recentlyObservedUnstableDuringInit());
+    }
+
+    @Test
+    public void stopping_edge_during_init_does_not_mark_as_unstable_init() {
+        ClusterFixture fixture = ClusterFixture.forFlatCluster(3).reportStorageNodeState(0, State.INITIALIZING);
+        fixture.reportStorageNodeState(0, State.STOPPING);
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 0));
+
+        assertFalse(nodeInfo.recentlyObservedUnstableDuringInit());
+    }
+
+    /**
+     * The cluster controller will, after a time of observed stable state, reset the crash
+     * counter for a given node. This should also reset the unstable init flag to keep it
+     * from haunting a now stable node.
+     */
+    @Test
+    public void zeroing_crash_count_resets_unstable_init_flag() {
+        ClusterFixture fixture = fixtureWithNodeMarkedAsUnstableInit(1);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1));
+        nodeInfo.setPrematureCrashCount(0);
+        assertFalse(nodeInfo.recentlyObservedUnstableDuringInit());
+    }
+
+    /**
+     * A non-zero crash count update, on the other hand, implies the node is suffering
+     * further instabilities and should not clear the unstable init flag.
+     */
+    @Test
+    public void non_zero_crash_count_update_does_not_reset_unstable_init_flag() {
+        ClusterFixture fixture = fixtureWithNodeMarkedAsUnstableInit(1);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1));
+        nodeInfo.setPrematureCrashCount(3);
+        assertTrue(nodeInfo.recentlyObservedUnstableDuringInit());
+    }
+
+    @Test
+    public void non_zero_crash_count_does_not_implicitly_set_unstable_init_flag() {
+        ClusterFixture fixture = ClusterFixture.forFlatCluster(3);
+
+        final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1));
+        nodeInfo.setPrematureCrashCount(1);
+        assertFalse(nodeInfo.recentlyObservedUnstableDuringInit());
+    }
+
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/RpcServerTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/RpcServerTest.java
index 2816b75622e..f7f86907205 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/RpcServerTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/RpcServerTest.java
@@ -437,13 +437,13 @@ public class RpcServerTest extends FleetControllerTest {
         { // Configuration change: Remove the previously retired nodes
             /*
             TODO: Verify current result: version:23 distributor:7 .0.s:d .1.s:d .2.s:d .3.s:d .4.s:d storage:7 .0.s:m .1.s:m .2.s:m .3.s:m .4.s:m
-            TODO: Make this work without stopping/disconnecting (see SystemStateGenerator.setNodes
+            TODO: Make this work without stopping/disconnecting (see StateChangeHandler.setNodes
             Set<ConfiguredNode> configuredNodes = new TreeSet<>();
             configuredNodes.add(new ConfiguredNode(5, false));
             configuredNodes.add(new ConfiguredNode(6, false));
             FleetControllerOptions options = new FleetControllerOptions("mycluster", configuredNodes);
             options.slobrokConnectionSpecs = this.options.slobrokConnectionSpecs;
-            this.options.maxInitProgressTime = 30000;
+            this.options.maxInitProgressTimeMs = 30000;
             this.options.stableStateTimePeriod = 60000;
             fleetController.updateOptions(options, 0);
             for (int i = 0; i < 5*2; i++) {
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/SystemStateGeneratorTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateChangeHandlerTest.java
index 35118933b42..f591e8efc06 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/SystemStateGeneratorTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateChangeHandlerTest.java
@@ -6,7 +6,6 @@ import com.yahoo.vdslib.distribution.Distribution;
 import com.yahoo.vdslib.state.*;
 import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo;
 import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler;
-import com.yahoo.vespa.clustercontroller.core.listeners.SystemStateListener;
 import com.yahoo.vespa.clustercontroller.core.mocks.TestEventLog;
 import com.yahoo.vespa.clustercontroller.core.testutils.LogFormatter;
 import junit.framework.TestCase;
@@ -16,33 +15,16 @@ import java.util.Set;
 import java.util.TreeSet;
 import java.util.logging.Logger;
 
-public class SystemStateGeneratorTest extends TestCase {
-    private static final Logger log = Logger.getLogger(SystemStateGeneratorTest.class.getName());
-    class Config {
+public class StateChangeHandlerTest extends TestCase {
+    private static final Logger log = Logger.getLogger(StateChangeHandlerTest.class.getName());
+    private class Config {
         int nodeCount = 3;
         int stableStateTime = 1000 * 60000;
         int maxSlobrokDisconnectPeriod = 60000;
         int maxPrematureCrashes = 3;
     }
-    class TestSystemStateListener implements SystemStateListener {
-        LinkedList<ClusterState> states = new LinkedList<>();
 
-        @Override
-        public void handleNewSystemState(ClusterState state) {
-            states.add(state);
-        }
-
-        public String toString() {
-            StringBuilder sb = new StringBuilder();
-            sb.append("States(");
-            for (ClusterState state : states) sb.append('\n').append(state.toString());
-            sb.append(")");
-            return sb.toString();
-        }
-
-    }
-
-    class TestNodeStateOrHostInfoChangeHandler implements NodeStateOrHostInfoChangeHandler {
+    private class TestNodeStateOrHostInfoChangeHandler implements NodeStateOrHostInfoChangeHandler {
 
         LinkedList<String> events = new LinkedList<>();
 
@@ -75,9 +57,9 @@ public class SystemStateGeneratorTest extends TestCase {
     private Set<ConfiguredNode> configuredNodes = new TreeSet<>();
     private Config config;
     private ContentCluster cluster;
-    private SystemStateGenerator generator;
-    private TestSystemStateListener systemStateListener;
+    private StateChangeHandler nodeStateChangeHandler;
     private TestNodeStateOrHostInfoChangeHandler nodeStateUpdateListener;
+    private final ClusterStateGenerator.Params params = new ClusterStateGenerator.Params();
 
     public void setUp() {
         LogFormatter.initializeLogging();
@@ -88,20 +70,18 @@ public class SystemStateGeneratorTest extends TestCase {
         this.config = config;
         for (int i=0; i<config.nodeCount; ++i) configuredNodes.add(new ConfiguredNode(i, false));
         cluster = new ContentCluster("testcluster", configuredNodes, distribution, 0, 0.0);
-        generator = new SystemStateGenerator(clock, eventLog, null);
-        generator.setNodes(cluster.clusterInfo());
-        generator.setStableStateTimePeriod(config.stableStateTime);
-        generator.setMaxPrematureCrashes(config.maxPrematureCrashes);
-        generator.setMaxSlobrokDisconnectGracePeriod(config.maxSlobrokDisconnectPeriod);
-        generator.setMinNodesUp(1, 1, 0, 0);
-        systemStateListener = new TestSystemStateListener();
+        nodeStateChangeHandler = new StateChangeHandler(clock, eventLog, null);
+        params.minStorageNodesUp(1).minDistributorNodesUp(1)
+                .minRatioOfStorageNodesUp(0.0).minRatioOfDistributorNodesUp(0.0)
+                .maxPrematureCrashes(config.maxPrematureCrashes)
+                .transitionTimes(5000)
+                .cluster(cluster);
         nodeStateUpdateListener = new TestNodeStateOrHostInfoChangeHandler();
     }
 
-    private void assertNewClusterStateReceived() {
-        assertTrue(generator.notifyIfNewSystemState(cluster, systemStateListener));
-        assertTrue(systemStateListener.toString(), systemStateListener.states.size() == 1);
-        systemStateListener.states.clear();
+    private ClusterState currentClusterState() {
+        params.currentTimeInMilllis(clock.getCurrentTimeInMillis());
+        return ClusterStateGenerator.generatedStateFrom(params).getClusterState();
     }
 
     private void startWithStableStateClusterWithNodesUp() {
@@ -109,61 +89,55 @@ public class SystemStateGeneratorTest extends TestCase {
             for (ConfiguredNode i : configuredNodes) {
                 NodeInfo nodeInfo = cluster.clusterInfo().setRpcAddress(new Node(type, i.index()), null);
                 nodeInfo.markRpcAddressLive();
-                generator.handleNewReportedNodeState(nodeInfo, new NodeState(type, State.UP), null);
+                nodeStateChangeHandler.handleNewReportedNodeState(
+                        currentClusterState(), nodeInfo, new NodeState(type, State.UP), null);
                 nodeInfo.setReportedState(new NodeState(type, State.UP), clock.getCurrentTimeInMillis());
             }
         }
-        assertNewClusterStateReceived();
         for (NodeType type : NodeType.getTypes()) {
             for (ConfiguredNode i : configuredNodes) {
                 Node n = new Node(type, i.index());
-                assertEquals(State.UP, generator.getClusterState().getNodeState(n).getState());
+                assertEquals(State.UP, currentClusterState().getNodeState(n).getState());
             }
         }
         clock.advanceTime(config.stableStateTime);
     }
 
     private void markNodeOutOfSlobrok(Node node) {
+        final ClusterState stateBefore = currentClusterState();
         log.info("Marking " + node + " out of slobrok");
         cluster.getNodeInfo(node).markRpcAddressOutdated(clock);
-        generator.handleMissingNode(cluster.getNodeInfo(node), nodeStateUpdateListener);
-        assertTrue(nodeStateUpdateListener.toString(), nodeStateUpdateListener.events.isEmpty());
-        nodeStateUpdateListener.events.clear();
+        nodeStateChangeHandler.handleMissingNode(stateBefore, cluster.getNodeInfo(node), nodeStateUpdateListener);
         assertTrue(eventLog.toString(), eventLog.toString().contains("Node is no longer in slobrok"));
         eventLog.clear();
     }
 
     private void markNodeBackIntoSlobrok(Node node, State state) {
+        final ClusterState stateBefore = currentClusterState();
         log.info("Marking " + node + " back in slobrok");
         cluster.getNodeInfo(node).markRpcAddressLive();
-        generator.handleReturnedRpcAddress(cluster.getNodeInfo(node));
-        assertEquals(0, nodeStateUpdateListener.events.size());
-        assertEquals(0, systemStateListener.states.size());
-        generator.handleNewReportedNodeState(cluster.getNodeInfo(node), new NodeState(node.getType(), state), nodeStateUpdateListener);
+        nodeStateChangeHandler.handleReturnedRpcAddress(cluster.getNodeInfo(node));
+        nodeStateChangeHandler.handleNewReportedNodeState(
+                stateBefore, cluster.getNodeInfo(node),
+                new NodeState(node.getType(), state), nodeStateUpdateListener);
         cluster.getNodeInfo(node).setReportedState(new NodeState(node.getType(), state), clock.getCurrentTimeInMillis());
-        assertEquals(0, nodeStateUpdateListener.events.size());
-        assertEquals(0, systemStateListener.states.size());
     }
 
     private void verifyClusterStateChanged(Node node, State state) {
         log.info("Verifying cluster state has been updated for " + node + " to " + state);
-        assertTrue(generator.notifyIfNewSystemState(cluster, systemStateListener));
-        assertEquals(1, systemStateListener.states.size());
-        assertEquals(state, systemStateListener.states.get(0).getNodeState(node).getState());
-        systemStateListener.states.clear();
-        assertEquals(state, generator.getClusterState().getNodeState(node).getState());
+        assertTrue(nodeStateChangeHandler.stateMayHaveChanged());
+        assertEquals(state, currentClusterState().getNodeState(node).getState());
     }
 
     private void verifyNodeStateAfterTimerWatch(Node node, State state) {
         log.info("Verifying state of node after timer watch.");
-        generator.watchTimers(cluster, nodeStateUpdateListener);
+        nodeStateChangeHandler.watchTimers(cluster, currentClusterState(), nodeStateUpdateListener);
         assertEquals(0, nodeStateUpdateListener.events.size());
         verifyClusterStateChanged(node, state);
     }
 
     private void verifyPrematureCrashCountCleared(Node node) {
-        assertTrue(generator.watchTimers(cluster, nodeStateUpdateListener));
-        assertEquals(0, nodeStateUpdateListener.events.size());
+        assertTrue(nodeStateChangeHandler.watchTimers(cluster, currentClusterState(), nodeStateUpdateListener));
         assertEquals(0, cluster.getNodeInfo(node).getPrematureCrashCount());
     }
 
@@ -175,15 +149,15 @@ public class SystemStateGeneratorTest extends TestCase {
             log.info("Iteration " + j);
             assertEquals(0, cluster.getNodeInfo(node).getPrematureCrashCount());
             assertEquals(State.UP, cluster.getNodeInfo(node).getWantedState().getState());
-            assertEquals(State.UP, generator.getClusterState().getNodeState(node).getState());
+            assertEquals(State.UP, currentClusterState().getNodeState(node).getState());
             for (int k=0; k<config.maxPrematureCrashes; ++k) {
                 log.info("Premature iteration " + k);
                 markNodeOutOfSlobrok(node);
 
                 log.info("Passing max disconnect time period. Watching timers");
                 clock.advanceTime(config.maxSlobrokDisconnectPeriod);
-
                 verifyNodeStateAfterTimerWatch(node, State.MAINTENANCE);
+
                 cluster.getNodeInfo(node).setReportedState(new NodeState(node.getType(), State.DOWN), clock.getCurrentTimeInMillis());
 
                 assertEquals(k, cluster.getNodeInfo(node).getPrematureCrashCount());
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateChangeTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateChangeTest.java
index b94691bb880..c31f80d9b53 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateChangeTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateChangeTest.java
@@ -8,8 +8,10 @@ import com.yahoo.vespa.clustercontroller.core.database.DatabaseHandler;
 import com.yahoo.vespa.clustercontroller.core.testutils.StateWaiter;
 import com.yahoo.vespa.clustercontroller.utils.util.NoMetricReporter;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
@@ -43,7 +45,7 @@ public class StateChangeTest extends FleetControllerTest {
                                                     options.minStorageNodesUp, options.minRatioOfStorageNodesUp);
         NodeStateGatherer stateGatherer = new NodeStateGatherer(timer, timer, eventLog);
         DatabaseHandler database = new DatabaseHandler(timer, options.zooKeeperServerAddress, options.fleetControllerIndex, timer);
-        SystemStateGenerator stateGenerator = new SystemStateGenerator(timer, eventLog, metricUpdater);
+        StateChangeHandler stateGenerator = new StateChangeHandler(timer, eventLog, metricUpdater);
         SystemStateBroadcaster stateBroadcaster = new SystemStateBroadcaster(timer, timer);
         MasterElectionHandler masterElectionHandler = new MasterElectionHandler(options.fleetControllerIndex, options.fleetControllerCount, timer, timer);
         ctrl = new FleetController(timer, eventLog, cluster, stateGatherer, communicator, null, null, communicator, database, stateGenerator, stateBroadcaster, masterElectionHandler, metricUpdater, options);
@@ -109,8 +111,13 @@ public class StateChangeTest extends FleetControllerTest {
         // Now, fleet controller should have generated a new cluster state.
         ctrl.tick();
 
-        assertEquals("version:6 distributor:10 .0.s:i .0.i:0.0 .1.s:i .1.i:0.0 .2.s:i .2.i:0.0 .3.s:i .3.i:0.0 .4.s:i .4.i:0.0 .5.s:i .5.i:0.0 .6.s:i .6.i:0.0 .7.s:i .7.i:0.0 .8.s:i .8.i:0.0 .9.s:i .9.i:0.0 storage:10 .0.s:i .0.i:0.9 .1.s:i .1.i:0.9 .2.s:i .2.i:0.9 .3.s:i .3.i:0.9 .4.s:i .4.i:0.9 .5.s:i .5.i:0.9 .6.s:i .6.i:0.9 .7.s:i .7.i:0.9 .8.s:i .8.i:0.9 .9.s:i .9.i:0.9",
-                     ctrl.getSystemState().toString());
+        // Regular init progress does not update the cluster state until the node is done initializing (or goes down,
+        // whichever comes first).
+        assertEquals("version:6 distributor:10 .0.s:i .0.i:0.0 .1.s:i .1.i:0.0 .2.s:i .2.i:0.0 .3.s:i .3.i:0.0 " +
+                        ".4.s:i .4.i:0.0 .5.s:i .5.i:0.0 .6.s:i .6.i:0.0 .7.s:i .7.i:0.0 .8.s:i .8.i:0.0 " +
+                        ".9.s:i .9.i:0.0 storage:10 .0.s:i .0.i:0.1 .1.s:i .1.i:0.1 .2.s:i .2.i:0.1 .3.s:i .3.i:0.1 " +
+                        ".4.s:i .4.i:0.1 .5.s:i .5.i:0.1 .6.s:i .6.i:0.1 .7.s:i .7.i:0.1 .8.s:i .8.i:0.1 .9.s:i .9.i:0.1",
+                ctrl.consolidatedClusterState().toString());
 
         timer.advanceTime(options.maxInitProgressTime / 20);
         ctrl.tick();
@@ -131,24 +138,23 @@ public class StateChangeTest extends FleetControllerTest {
 
         assertEquals("version:8 distributor:10 storage:10", ctrl.getSystemState().toString());
 
-
         verifyNodeEvents(new Node(NodeType.DISTRIBUTOR, 0),
                 "Event: distributor.0: Now reporting state U\n" +
-                "Event: distributor.0: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: distributor.0: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: distributor.0: Now reporting state I, i 0.00\n" +
-                "Event: distributor.0: Altered node state in cluster state from 'U' to 'I, i 0.00'.\n" +
+                "Event: distributor.0: Altered node state in cluster state from 'U' to 'I, i 0.00'\n" +
                 "Event: distributor.0: Now reporting state U\n" +
-                "Event: distributor.0: Altered node state in cluster state from 'I, i 0.00' to 'U'.\n");
+                "Event: distributor.0: Altered node state in cluster state from 'I, i 0.00' to 'U'\n");
 
         verifyNodeEvents(new Node(NodeType.STORAGE, 0),
                 "Event: storage.0: Now reporting state U\n" +
-                "Event: storage.0: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: storage.0: Now reporting state I, i 0.00 (ls)\n" +
-                "Event: storage.0: Altered node state in cluster state from 'U' to 'D: Listing buckets. Progress 0.0 %.'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'U' to 'D'\n" +
                 "Event: storage.0: Now reporting state I, i 0.100 (read)\n" +
-                "Event: storage.0: Altered node state in cluster state from 'D: Listing buckets. Progress 0.0 %.' to 'I, i 0.100 (read)'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'D' to 'I, i 0.100 (read)'\n" +
                 "Event: storage.0: Now reporting state U\n" +
-                "Event: storage.0: Altered node state in cluster state from 'I, i 0.900 (read)' to 'U'.\n");
+                "Event: storage.0: Altered node state in cluster state from 'I, i 0.100 (read)' to 'U'\n");
     }
 
     @Test
@@ -172,7 +178,6 @@ public class StateChangeTest extends FleetControllerTest {
         assertEquals("version:4 distributor:10 .0.s:d storage:10", ctrl.getSystemState().toString());
 
         timer.advanceTime(1000);
-        long distStartTime = timer.getCurrentTimeInMillis() / 1000;
 
         ctrl.tick();
 
@@ -210,23 +215,24 @@ public class StateChangeTest extends FleetControllerTest {
 
         verifyNodeEvents(new Node(NodeType.DISTRIBUTOR, 0),
                 "Event: distributor.0: Now reporting state U\n" +
-                "Event: distributor.0: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: distributor.0: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: distributor.0: Failed to get node state: D: Closed at other end\n" +
                 "Event: distributor.0: Stopped or possibly crashed after 0 ms, which is before stable state time period. Premature crash count is now 1.\n" +
-                "Event: distributor.0: Altered node state in cluster state from 'U' to 'D: Closed at other end'.\n" +
+                "Event: distributor.0: Altered node state in cluster state from 'U' to 'D: Closed at other end'\n" +
                 "Event: distributor.0: Now reporting state U, t 12345678\n" +
-                "Event: distributor.0: Altered node state in cluster state from 'D: Closed at other end' to 'U, t 12345678'.\n");
+                "Event: distributor.0: Altered node state in cluster state from 'D: Closed at other end' to 'U, t 12345678'\n" +
+                "Event: distributor.0: Altered node state in cluster state from 'U, t 12345678' to 'U'\n");
 
         verifyNodeEvents(new Node(NodeType.STORAGE, 0),
                 "Event: storage.0: Now reporting state U\n" +
-                "Event: storage.0: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: storage.0: Failed to get node state: D: Closed at other end\n" +
                 "Event: storage.0: Stopped or possibly crashed after 1000 ms, which is before stable state time period. Premature crash count is now 1.\n" +
-                "Event: storage.0: Altered node state in cluster state from 'U' to 'M: Closed at other end'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'U' to 'M: Closed at other end'\n" +
                 "Event: storage.0: 5001 milliseconds without contact. Marking node down.\n" +
-                "Event: storage.0: Altered node state in cluster state from 'M: Closed at other end' to 'D: Closed at other end'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'M: Closed at other end' to 'D: Closed at other end'\n" +
                 "Event: storage.0: Now reporting state U, t 12345679\n" +
-                "Event: storage.0: Altered node state in cluster state from 'D: Closed at other end' to 'U, t 12345679'.\n");
+                "Event: storage.0: Altered node state in cluster state from 'D: Closed at other end' to 'U, t 12345679'\n");
 
         assertEquals(1, ctrl.getCluster().getNodeInfo(new Node(NodeType.DISTRIBUTOR, 0)).getPrematureCrashCount());
         assertEquals(1, ctrl.getCluster().getNodeInfo(new Node(NodeType.STORAGE, 0)).getPrematureCrashCount());
@@ -239,7 +245,7 @@ public class StateChangeTest extends FleetControllerTest {
 
     @Test
     public void testNodeGoingDownAndUpNotifying() throws Exception {
-        // Same test as above, but node manage to notify why it is going down first.
+        // Same test as above, but node manages to notify why it is going down first.
         FleetControllerOptions options = new FleetControllerOptions("mycluster", createNodes(10));
         options.nodeStateRequestTimeoutMS = 60 * 60 * 1000;
         options.maxSlobrokDisconnectGracePeriod = 100000;
@@ -291,21 +297,21 @@ public class StateChangeTest extends FleetControllerTest {
 
         verifyNodeEvents(new Node(NodeType.DISTRIBUTOR, 0),
                 "Event: distributor.0: Now reporting state U\n" +
-                "Event: distributor.0: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: distributor.0: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: distributor.0: Failed to get node state: D: controlled shutdown\n" +
-                "Event: distributor.0: Altered node state in cluster state from 'U' to 'D: controlled shutdown'.\n" +
+                "Event: distributor.0: Altered node state in cluster state from 'U' to 'D: controlled shutdown'\n" +
                 "Event: distributor.0: Now reporting state U\n" +
-                "Event: distributor.0: Altered node state in cluster state from 'D: controlled shutdown' to 'U'.\n");
+                "Event: distributor.0: Altered node state in cluster state from 'D: controlled shutdown' to 'U'\n");
 
         verifyNodeEvents(new Node(NodeType.STORAGE, 0),
                 "Event: storage.0: Now reporting state U\n" +
-                "Event: storage.0: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: storage.0: Failed to get node state: D: controlled shutdown\n" +
-                "Event: storage.0: Altered node state in cluster state from 'U' to 'M: controlled shutdown'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'U' to 'M: controlled shutdown'\n" +
                 "Event: storage.0: 5001 milliseconds without contact. Marking node down.\n" +
-                "Event: storage.0: Altered node state in cluster state from 'M: controlled shutdown' to 'D: controlled shutdown'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'M: controlled shutdown' to 'D: controlled shutdown'\n" +
                 "Event: storage.0: Now reporting state U\n" +
-                "Event: storage.0: Altered node state in cluster state from 'D: controlled shutdown' to 'U'.\n");
+                "Event: storage.0: Altered node state in cluster state from 'D: controlled shutdown' to 'U'\n");
 
     }
 
@@ -346,7 +352,7 @@ public class StateChangeTest extends FleetControllerTest {
 
         verifyNodeEvents(new Node(NodeType.STORAGE, 0),
                 "Event: storage.0: Now reporting state U\n" +
-                "Event: storage.0: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: storage.0: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: storage.0: Node is no longer in slobrok, but we still have a pending state request.\n");
     }
 
@@ -393,15 +399,15 @@ public class StateChangeTest extends FleetControllerTest {
 
         verifyNodeEvents(new Node(NodeType.STORAGE, 6),
                 "Event: storage.6: Now reporting state U\n" +
-                "Event: storage.6: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: storage.6: Failed to get node state: D: Connection error: Closed at other end\n" +
                 "Event: storage.6: Stopped or possibly crashed after 0 ms, which is before stable state time period. Premature crash count is now 1.\n" +
-                "Event: storage.6: Altered node state in cluster state from 'U' to 'M: Connection error: Closed at other end'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'U' to 'M: Connection error: Closed at other end'\n" +
                 "Event: storage.6: Now reporting state I, i 0.00 (ls)\n" +
                 "Event: storage.6: Now reporting state I, i 0.600 (read)\n" +
-                "Event: storage.6: Altered node state in cluster state from 'M: Connection error: Closed at other end' to 'I, i 0.600 (read)'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'M: Connection error: Closed at other end' to 'I, i 0.600 (read)'\n" +
                 "Event: storage.6: Now reporting state U\n" +
-                "Event: storage.6: Altered node state in cluster state from 'I, i 0.600 (read)' to 'U'.\n");
+                "Event: storage.6: Altered node state in cluster state from 'I, i 0.600 (read)' to 'U'\n");
     }
 
     @Test
@@ -453,14 +459,14 @@ public class StateChangeTest extends FleetControllerTest {
 
         verifyNodeEvents(new Node(NodeType.STORAGE, 6),
                 "Event: storage.6: Now reporting state U\n" +
-                "Event: storage.6: Altered node state in cluster state from 'D' to 'R'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'R'\n" +
                 "Event: storage.6: Failed to get node state: D: Connection error: Closed at other end\n" +
                 "Event: storage.6: Stopped or possibly crashed after 0 ms, which is before stable state time period. Premature crash count is now 1.\n" +
-                "Event: storage.6: Altered node state in cluster state from 'R' to 'M: Connection error: Closed at other end'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'R' to 'M: Connection error: Closed at other end'\n" +
                 "Event: storage.6: Now reporting state I, i 0.00 (ls)\n" +
                 "Event: storage.6: Now reporting state I, i 0.600 (read)\n" +
                 "Event: storage.6: Now reporting state U\n" +
-                "Event: storage.6: Altered node state in cluster state from 'M: Connection error: Closed at other end' to 'R: Connection error: Closed at other end'.\n");
+                "Event: storage.6: Altered node state in cluster state from 'M: Connection error: Closed at other end' to 'R'\n");
     }
 
     @Test
@@ -522,7 +528,7 @@ public class StateChangeTest extends FleetControllerTest {
 
         ctrl.tick();
 
-        assertEquals("Listing buckets. Progress 0.1 %.", ctrl.getSystemState().getNodeState(new Node(NodeType.STORAGE, 6)).getDescription());
+        assertEquals("version:5 distributor:10 storage:10 .6.s:d", ctrl.getSystemState().toString());
 
         communicator.setNodeState(new Node(NodeType.STORAGE, 6), new NodeState(NodeType.STORAGE, State.INITIALIZING).setInitProgress(0.1), "");
 
@@ -542,16 +548,16 @@ public class StateChangeTest extends FleetControllerTest {
 
         verifyNodeEvents(new Node(NodeType.STORAGE, 6),
                 "Event: storage.6: Now reporting state U\n" +
-                "Event: storage.6: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: storage.6: Failed to get node state: D: Connection error: Closed at other end\n" +
-                "Event: storage.6: Altered node state in cluster state from 'U' to 'M: Connection error: Closed at other end'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'U' to 'M: Connection error: Closed at other end'\n" +
                 "Event: storage.6: 100000 milliseconds without contact. Marking node down.\n" +
-                "Event: storage.6: Altered node state in cluster state from 'M: Connection error: Closed at other end' to 'D: Connection error: Closed at other end'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'M: Connection error: Closed at other end' to 'D: Connection error: Closed at other end'\n" +
                 "Event: storage.6: Now reporting state I, i 0.00100 (ls)\n" +
                 "Event: storage.6: Now reporting state I, i 0.100 (read)\n" +
-                "Event: storage.6: Altered node state in cluster state from 'D: Listing buckets. Progress 0.1 %.' to 'I, i 0.100 (read)'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'D: Connection error: Closed at other end' to 'I, i 0.100 (read)'\n" +
                 "Event: storage.6: Now reporting state U\n" +
-                "Event: storage.6: Altered node state in cluster state from 'I, i 0.100 (read)' to 'U'.\n");
+                "Event: storage.6: Altered node state in cluster state from 'I, i 0.100 (read)' to 'U'\n");
     }
 
     @Test
@@ -613,9 +619,6 @@ public class StateChangeTest extends FleetControllerTest {
         // Still down since it seemingly crashed during last init.
         assertEquals("version:7 distributor:10 storage:10 .6.s:d", ctrl.getSystemState().toString());
 
-        assertEquals("Down: 5001 ms without initialize progress. Assuming node has deadlocked.",
-                ctrl.getSystemState().getNodeState(new Node(NodeType.STORAGE, 6)).toString());
-
         ctrl.tick();
 
         communicator.setNodeState(new Node(NodeType.STORAGE, 6), State.UP, "");
@@ -626,20 +629,20 @@ public class StateChangeTest extends FleetControllerTest {
 
         verifyNodeEvents(new Node(NodeType.STORAGE, 6),
                 "Event: storage.6: Now reporting state U\n" +
-                "Event: storage.6: Altered node state in cluster state from 'D' to 'U'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
                 "Event: storage.6: Failed to get node state: D: Connection error: Closed at other end\n" +
-                "Event: storage.6: Altered node state in cluster state from 'U' to 'M: Connection error: Closed at other end'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'U' to 'M: Connection error: Closed at other end'\n" +
                 "Event: storage.6: 1000000 milliseconds without contact. Marking node down.\n" +
-                "Event: storage.6: Altered node state in cluster state from 'M: Connection error: Closed at other end' to 'D: Connection error: Closed at other end'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'M: Connection error: Closed at other end' to 'D: Connection error: Closed at other end'\n" +
                 "Event: storage.6: Now reporting state I, i 0.100 (read)\n" +
-                "Event: storage.6: Altered node state in cluster state from 'D: Connection error: Closed at other end' to 'I, i 0.100 (read)'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'D: Connection error: Closed at other end' to 'I, i 0.100 (read)'\n" +
                 "Event: storage.6: 5001 milliseconds without initialize progress. Marking node down. Premature crash count is now 1.\n" +
-                "Event: storage.6: Altered node state in cluster state from 'I, i 0.100 (read)' to 'D: 5001 ms without initialize progress. Assuming node has deadlocked.'.\n" +
+                "Event: storage.6: Altered node state in cluster state from 'I, i 0.100 (read)' to 'D'\n" +
                 "Event: storage.6: Failed to get node state: D: Connection error: Closed at other end\n" +
                 "Event: storage.6: Now reporting state I, i 0.00 (ls)\n" +
                 "Event: storage.6: Now reporting state I, i 0.100 (read)\n" +
                 "Event: storage.6: Now reporting state U\n" +
-                "Event: storage.6: Altered node state in cluster state from 'D: 5001 ms without initialize progress. Assuming node has deadlocked.' to 'U'.\n");
+                "Event: storage.6: Altered node state in cluster state from 'D' to 'U'\n");
 
     }
 
@@ -684,9 +687,6 @@ public class StateChangeTest extends FleetControllerTest {
         ctrl.tick();
 
         assertEquals("version:7 distributor:10 storage:10 .6.s:d", ctrl.getSystemState().toString());
-
-        String desc = ctrl.getSystemState().getNodeState(new Node(NodeType.STORAGE, 6)).getDescription();
-        assertEquals("Got reverse intialize progress. Assuming node have prematurely crashed", desc);
     }
 
     @Test
@@ -1132,4 +1132,70 @@ public class StateChangeTest extends FleetControllerTest {
         }
     }
 
+    @Test
+    public void consolidated_cluster_state_reflects_node_changes_when_cluster_is_down() throws Exception {
+        FleetControllerOptions options = new FleetControllerOptions("mycluster", createNodes(10));
+        options.maxTransitionTime.put(NodeType.STORAGE, 0);
+        options.minStorageNodesUp = 10;
+        options.minDistributorNodesUp = 10;
+        initialize(options);
+
+        ctrl.tick();
+        assertThat(ctrl.consolidatedClusterState().toString(), equalTo("version:3 distributor:10 storage:10"));
+
+        communicator.setNodeState(new Node(NodeType.STORAGE, 2), State.DOWN, "foo");
+        ctrl.tick();
+
+        assertThat(ctrl.consolidatedClusterState().toString(),
+                   equalTo("version:4 cluster:d distributor:10 storage:10 .2.s:d"));
+
+        // After this point, any further node changes while the cluster is still down won't be published.
+        // This is because cluster state similarity checks are short-circuited if both are Down, as no other parts
+        // of the state matter. Despite this, REST API access and similar features need up-to-date information,
+        // and therefore need to get a state which represents the _current_ state rather than the published state.
+        // The consolidated state offers this by selectively generating the current state on-demand if the
+        // cluster is down.
+        communicator.setNodeState(new Node(NodeType.STORAGE, 5), State.DOWN, "bar");
+        ctrl.tick();
+
+        // NOTE: _same_ version, different node state content. Overall cluster down-state is still the same.
+        assertThat(ctrl.consolidatedClusterState().toString(),
+                   equalTo("version:4 cluster:d distributor:10 storage:10 .2.s:d .5.s:d"));
+    }
+
+    // Related to the above test, watchTimer invocations must receive the _current_ state and not the
+    // published state. Failure to ensure this would cause events to be fired non-stop, as the effect
+    // of previous timer invocations (with subsequent state generation) would not be visible.
+    @Test
+    public void timer_events_during_cluster_down_observe_most_recent_node_changes() throws Exception {
+        FleetControllerOptions options = new FleetControllerOptions("mycluster", createNodes(10));
+        options.maxTransitionTime.put(NodeType.STORAGE, 1000);
+        options.minStorageNodesUp = 10;
+        options.minDistributorNodesUp = 10;
+        initialize(options);
+
+        ctrl.tick();
+        communicator.setNodeState(new Node(NodeType.STORAGE, 2), State.DOWN, "foo");
+        timer.advanceTime(500);
+        ctrl.tick();
+        communicator.setNodeState(new Node(NodeType.STORAGE, 3), State.DOWN, "foo");
+        ctrl.tick();
+        assertThat(ctrl.consolidatedClusterState().toString(), equalTo("version:4 cluster:d distributor:10 storage:10 .2.s:m .3.s:m"));
+
+        // Subsequent timer tick should _not_ trigger additional events. Providing published state
+        // only would result in "Marking node down" events for node 2 emitted per tick.
+        for (int i = 0; i < 3; ++i) {
+            timer.advanceTime(5000);
+            ctrl.tick();
+        }
+
+        verifyNodeEvents(new Node(NodeType.STORAGE, 2),
+                "Event: storage.2: Now reporting state U\n" +
+                "Event: storage.2: Altered node state in cluster state from 'D: Node not seen in slobrok.' to 'U'\n" +
+                "Event: storage.2: Failed to get node state: D: foo\n" +
+                "Event: storage.2: Stopped or possibly crashed after 500 ms, which is before stable state time period. Premature crash count is now 1.\n" +
+                "Event: storage.2: Altered node state in cluster state from 'U' to 'M: foo'\n" +
+                "Event: storage.2: 5000 milliseconds without contact. Marking node down.\n");
+    }
+
 }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java
new file mode 100644
index 00000000000..72f8c9fb8b7
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java
@@ -0,0 +1,229 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+import com.yahoo.vdslib.state.ClusterState;
+import com.yahoo.vdslib.state.Node;
+import com.yahoo.vdslib.state.NodeState;
+import com.yahoo.vdslib.state.NodeType;
+import com.yahoo.vdslib.state.State;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Optional;
+
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.core.IsEqual.equalTo;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+public class StateVersionTrackerTest {
+
+    private static AnnotatedClusterState stateWithoutAnnotations(String stateStr) {
+        final ClusterState state = ClusterState.stateFromString(stateStr);
+        return new AnnotatedClusterState(state, Optional.empty(), AnnotatedClusterState.emptyNodeStateReasons());
+    }
+
+    private static StateVersionTracker createWithMockedMetrics() {
+        return new StateVersionTracker(mock(MetricUpdater.class));
+    }
+
+    private static void updateAndPromote(final StateVersionTracker versionTracker,
+                                         final AnnotatedClusterState state,
+                                         final long timeMs)
+    {
+        versionTracker.updateLatestCandidateState(state);
+        versionTracker.promoteCandidateToVersionedState(timeMs);
+    }
+
+    @Test
+    public void version_is_incremented_when_new_state_is_applied() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        versionTracker.setVersionRetrievedFromZooKeeper(100);
+        updateAndPromote(versionTracker, stateWithoutAnnotations("distributor:2 storage:2"), 123);
+        assertThat(versionTracker.getCurrentVersion(), equalTo(101));
+        assertThat(versionTracker.getVersionedClusterState().toString(), equalTo("version:101 distributor:2 storage:2"));
+    }
+
+    @Test
+    public void version_is_1_upon_construction() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        assertThat(versionTracker.getCurrentVersion(), equalTo(1));
+    }
+
+    @Test
+    public void set_current_version_caps_lowest_version_to_1() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        versionTracker.setVersionRetrievedFromZooKeeper(0);
+        assertThat(versionTracker.getCurrentVersion(), equalTo(1));
+    }
+
+    @Test
+    public void new_version_from_zk_predicate_initially_false() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        assertThat(versionTracker.hasReceivedNewVersionFromZooKeeper(), is(false));
+    }
+
+    @Test
+    public void new_version_from_zk_predicate_true_after_setting_zk_version() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        versionTracker.setVersionRetrievedFromZooKeeper(5);
+        assertThat(versionTracker.hasReceivedNewVersionFromZooKeeper(), is(true));
+    }
+
+    @Test
+    public void new_version_from_zk_predicate_false_after_applying_higher_version() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        versionTracker.setVersionRetrievedFromZooKeeper(5);
+        updateAndPromote(versionTracker, stateWithoutAnnotations("distributor:2 storage:2"), 123);
+        assertThat(versionTracker.hasReceivedNewVersionFromZooKeeper(), is(false));
+    }
+
+    @Test
+    public void exposed_states_are_empty_upon_construction() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        assertThat(versionTracker.getVersionedClusterState().toString(), equalTo(""));
+        assertThat(versionTracker.getAnnotatedVersionedClusterState().getClusterState().toString(), equalTo(""));
+    }
+
+    @Test
+    public void diff_from_initial_state_implies_changed_state() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        versionTracker.updateLatestCandidateState(stateWithoutAnnotations("cluster:d"));
+        assertTrue(versionTracker.candidateChangedEnoughFromCurrentToWarrantPublish());
+    }
+
+    private static boolean stateChangedBetween(String fromState, String toState) {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        updateAndPromote(versionTracker, stateWithoutAnnotations(fromState), 123);
+        versionTracker.updateLatestCandidateState(stateWithoutAnnotations(toState));
+        return versionTracker.candidateChangedEnoughFromCurrentToWarrantPublish();
+    }
+
+    @Test
+    public void version_mismatch_not_counted_as_changed_state() {
+        assertFalse(stateChangedBetween("distributor:2 storage:2", "distributor:2 storage:2"));
+    }
+
+    @Test
+    public void different_distributor_node_count_implies_changed_state() {
+        assertTrue(stateChangedBetween("distributor:2 storage:2", "distributor:3 storage:2"));
+        assertTrue(stateChangedBetween("distributor:3 storage:2", "distributor:2 storage:2"));
+    }
+
+    @Test
+    public void different_storage_node_count_implies_changed_state() {
+        assertTrue(stateChangedBetween("distributor:2 storage:2", "distributor:2 storage:3"));
+        assertTrue(stateChangedBetween("distributor:2 storage:3", "distributor:2 storage:2"));
+    }
+
+    @Test
+    public void different_distributor_node_state_implies_changed_state() {
+        assertTrue(stateChangedBetween("distributor:2 storage:2", "distributor:2 .0.s:d storage:2"));
+        assertTrue(stateChangedBetween("distributor:2 .0.s:d storage:2", "distributor:2 storage:2"));
+    }
+
+    @Test
+    public void different_storage_node_state_implies_changed_state() {
+        assertTrue(stateChangedBetween("distributor:2 storage:2", "distributor:2 storage:2 .0.s:d"));
+        assertTrue(stateChangedBetween("distributor:2 storage:2 .0.s:d", "distributor:2 storage:2"));
+    }
+
+    @Test
+    public void lowest_observed_distribution_bit_is_initially_16() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        assertThat(versionTracker.getLowestObservedDistributionBits(), equalTo(16));
+    }
+
+    @Test
+    public void lowest_observed_distribution_bit_is_tracked_across_states() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        updateAndPromote(versionTracker, stateWithoutAnnotations("bits:15 distributor:2 storage:2"), 100);
+        assertThat(versionTracker.getLowestObservedDistributionBits(), equalTo(15));
+
+        updateAndPromote(versionTracker, stateWithoutAnnotations("bits:17 distributor:2 storage:2"), 200);
+        assertThat(versionTracker.getLowestObservedDistributionBits(), equalTo(15));
+
+        updateAndPromote(versionTracker, stateWithoutAnnotations("bits:14 distributor:2 storage:2"), 300);
+        assertThat(versionTracker.getLowestObservedDistributionBits(), equalTo(14));
+    }
+
+    // For similarity purposes, only the cluster-wide bits matter, not the individual node state
+    // min used bits. The former is derived from the latter, but the latter is not visible in the
+    // published state (but _is_ visible in the internal ClusterState structures).
+    @Test
+    public void per_node_min_bits_changes_are_not_considered_different() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        final AnnotatedClusterState stateWithMinBits = stateWithoutAnnotations("distributor:2 storage:2");
+        stateWithMinBits.getClusterState().setNodeState(
+                new Node(NodeType.STORAGE, 0),
+                new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(15));
+        updateAndPromote(versionTracker, stateWithMinBits, 123);
+        versionTracker.updateLatestCandidateState(stateWithoutAnnotations("distributor:2 storage:2"));
+        assertFalse(versionTracker.candidateChangedEnoughFromCurrentToWarrantPublish());
+    }
+
+    @Test
+    public void state_history_is_initially_empty() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        assertTrue(versionTracker.getClusterStateHistory().isEmpty());
+    }
+
+    private static ClusterStateHistoryEntry historyEntry(final String state, final long time) {
+        return new ClusterStateHistoryEntry(ClusterState.stateFromString(state), time);
+    }
+
+    @Test
+    public void applying_state_adds_to_cluster_state_history() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        updateAndPromote(versionTracker, stateWithoutAnnotations("distributor:2 storage:2") ,100);
+        updateAndPromote(versionTracker, stateWithoutAnnotations("distributor:3 storage:3"), 200);
+        updateAndPromote(versionTracker, stateWithoutAnnotations("distributor:4 storage:4"), 300);
+
+        // Note: newest entry first
+        assertThat(versionTracker.getClusterStateHistory(),
+                equalTo(Arrays.asList(
+                         historyEntry("version:4 distributor:4 storage:4", 300),
+                         historyEntry("version:3 distributor:3 storage:3", 200),
+                         historyEntry("version:2 distributor:2 storage:2", 100))));
+    }
+
+    @Test
+    public void old_states_pruned_when_state_history_limit_reached() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+        versionTracker.setMaxHistoryEntryCount(2);
+
+        updateAndPromote(versionTracker, stateWithoutAnnotations("distributor:2 storage:2") ,100);
+        updateAndPromote(versionTracker, stateWithoutAnnotations("distributor:3 storage:3"), 200);
+        updateAndPromote(versionTracker, stateWithoutAnnotations("distributor:4 storage:4"), 300);
+
+        assertThat(versionTracker.getClusterStateHistory(),
+                equalTo(Arrays.asList(
+                        historyEntry("version:4 distributor:4 storage:4", 300),
+                        historyEntry("version:3 distributor:3 storage:3", 200))));
+
+        updateAndPromote(versionTracker, stateWithoutAnnotations("distributor:5 storage:5"), 400);
+
+        assertThat(versionTracker.getClusterStateHistory(),
+                equalTo(Arrays.asList(
+                        historyEntry("version:5 distributor:5 storage:5", 400),
+                        historyEntry("version:4 distributor:4 storage:4", 300))));
+    }
+
+    @Test
+    public void can_get_latest_non_published_candidate_state() {
+        final StateVersionTracker versionTracker = createWithMockedMetrics();
+
+        AnnotatedClusterState candidate = stateWithoutAnnotations("distributor:2 storage:2");
+        versionTracker.updateLatestCandidateState(candidate);
+        assertThat(versionTracker.getLatestCandidateState(), equalTo(candidate));
+
+        candidate = stateWithoutAnnotations("distributor:3 storage:3");
+        versionTracker.updateLatestCandidateState(candidate);
+        assertThat(versionTracker.getLatestCandidateState(), equalTo(candidate));
+    }
+
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/ClusterEventWithDescription.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/ClusterEventWithDescription.java
new file mode 100644
index 00000000000..111a2c63144
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/ClusterEventWithDescription.java
@@ -0,0 +1,40 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core.matchers;
+
+import com.yahoo.vespa.clustercontroller.core.ClusterEvent;
+import com.yahoo.vespa.clustercontroller.core.NodeEvent;
+import org.hamcrest.Description;
+import org.hamcrest.Factory;
+import org.mockito.ArgumentMatcher;
+
+public class ClusterEventWithDescription extends ArgumentMatcher<ClusterEvent> {
+    private final String expected;
+
+    public ClusterEventWithDescription(String expected) {
+        this.expected = expected;
+    }
+
+    @Override
+    public boolean matches(Object o) {
+        if (!(o instanceof ClusterEvent)) {
+            return false;
+        }
+        return expected.equals(((ClusterEvent) o).getDescription());
+    }
+
+    @Override
+    public void describeTo(Description description) {
+        description.appendText(String.format("ClusterEvent with description '%s'", expected));
+    }
+
+    @Override
+    public void describeMismatch(Object item, Description description) {
+        ClusterEvent other = (ClusterEvent)item;
+        description.appendText(String.format("got description '%s'", other.getDescription()));
+    }
+
+    @Factory
+    public static ClusterEventWithDescription clusterEventWithDescription(String description) {
+        return new ClusterEventWithDescription(description);
+    }
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/EventForNode.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/EventForNode.java
new file mode 100644
index 00000000000..1f2372dea29
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/EventForNode.java
@@ -0,0 +1,37 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core.matchers;
+
+import com.yahoo.vdslib.state.Node;
+import com.yahoo.vespa.clustercontroller.core.NodeEvent;
+import org.hamcrest.Description;
+import org.hamcrest.Factory;
+import org.mockito.ArgumentMatcher;
+
+public class EventForNode extends ArgumentMatcher<NodeEvent> {
+    private final Node expected;
+
+    EventForNode(Node expected) {
+        this.expected = expected;
+    }
+
+    @Override
+    public boolean matches(Object o) {
+        return ((NodeEvent)o).getNode().getNode().equals(expected);
+    }
+
+    @Override
+    public void describeTo(Description description) {
+        description.appendText(String.format("NodeEvent for node %s", expected));
+    }
+
+    @Override
+    public void describeMismatch(Object item, Description description) {
+        NodeEvent other = (NodeEvent)item;
+        description.appendText(String.format("got node %s", other.getNode().getNode()));
+    }
+
+    @Factory
+    public static EventForNode eventForNode(Node expected) {
+        return new EventForNode(expected);
+    }
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/EventTimeIs.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/EventTimeIs.java
new file mode 100644
index 00000000000..c99505d28ee
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/EventTimeIs.java
@@ -0,0 +1,40 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core.matchers;
+
+import com.yahoo.vespa.clustercontroller.core.Event;
+import org.hamcrest.Description;
+import org.hamcrest.Factory;
+import org.mockito.ArgumentMatcher;
+
+public class EventTimeIs extends ArgumentMatcher<Event> {
+    private final long expected;
+
+    public EventTimeIs(long expected) {
+        this.expected = expected;
+    }
+
+    @Override
+    public boolean matches(Object o) {
+        if (!(o instanceof Event)) {
+            return false;
+        }
+        return expected == ((Event)o).getTimeMs();
+    }
+
+    @Override
+    public void describeTo(Description description) {
+        description.appendText(String.format("Event with time %d", expected));
+    }
+
+    @Override
+    public void describeMismatch(Object item, Description description) {
+        Event other = (Event)item;
+        description.appendText(String.format("event time is %d", other.getTimeMs()));
+    }
+
+    @Factory
+    public static EventTimeIs eventTimeIs(long time) {
+        return new EventTimeIs(time);
+    }
+}
+
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/EventTypeIs.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/EventTypeIs.java
new file mode 100644
index 00000000000..5430bc5d8a3
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/EventTypeIs.java
@@ -0,0 +1,27 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core.matchers;
+
+import com.yahoo.vespa.clustercontroller.core.NodeEvent;
+import org.hamcrest.Factory;
+import org.mockito.ArgumentMatcher;
+
+public class EventTypeIs extends ArgumentMatcher<NodeEvent> {
+    private final NodeEvent.Type expected;
+
+    public EventTypeIs(NodeEvent.Type expected) {
+        this.expected = expected;
+    }
+
+    @Override
+    public boolean matches(Object o) {
+        if (!(o instanceof NodeEvent)) {
+            return false;
+        }
+        return expected.equals(((NodeEvent)o).getType());
+    }
+
+    @Factory
+    public static EventTypeIs eventTypeIs(NodeEvent.Type type) {
+        return new EventTypeIs(type);
+    }
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/HasStateReasonForNode.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/HasStateReasonForNode.java
new file mode 100644
index 00000000000..a147b9af466
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/HasStateReasonForNode.java
@@ -0,0 +1,49 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core.matchers;
+
+import com.yahoo.vdslib.state.Node;
+import com.yahoo.vespa.clustercontroller.core.NodeStateReason;
+import org.hamcrest.Description;
+import org.hamcrest.Factory;
+import org.mockito.ArgumentMatcher;
+
+import java.util.Map;
+
+public class HasStateReasonForNode extends ArgumentMatcher<Map<Node, NodeStateReason>> {
+    private final Node node;
+    private final NodeStateReason expected;
+
+    public HasStateReasonForNode(Node node, NodeStateReason expected) {
+        this.node = node;
+        this.expected = expected;
+    }
+
+    @Override
+    public boolean matches(Object o) {
+        if (o == null || !(o instanceof Map)) {
+            return false;
+        }
+        return expected == ((Map)o).get(node);
+    }
+
+    @Override
+    public void describeTo(Description description) {
+        description.appendText(String.format("has node state reason %s", expected.toString()));
+    }
+
+    @Override
+    public void describeMismatch(Object item, Description description) {
+        @SuppressWarnings("unchecked")
+        Map<Node, NodeStateReason> other = (Map<Node, NodeStateReason>)item;
+        if (other.containsKey(node)) {
+            description.appendText(String.format("has reason %s", other.get(node).toString()));
+        } else {
+            description.appendText("has no entry for node");
+        }
+    }
+
+    @Factory
+    public static HasStateReasonForNode hasStateReasonForNode(Node node, NodeStateReason reason) {
+        return new HasStateReasonForNode(node, reason);
+    }
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/NodeEventWithDescription.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/NodeEventWithDescription.java
new file mode 100644
index 00000000000..5ac89030c23
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/matchers/NodeEventWithDescription.java
@@ -0,0 +1,39 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core.matchers;
+
+import com.yahoo.vespa.clustercontroller.core.NodeEvent;
+import org.hamcrest.Description;
+import org.hamcrest.Factory;
+import org.mockito.ArgumentMatcher;
+
+public class NodeEventWithDescription extends ArgumentMatcher<NodeEvent> {
+    private final String expected;
+
+    public NodeEventWithDescription(String expected) {
+        this.expected = expected;
+    }
+
+    @Override
+    public boolean matches(Object o) {
+        if (!(o instanceof NodeEvent)) {
+            return false;
+        }
+        return expected.equals(((NodeEvent) o).getDescription());
+    }
+
+    @Override
+    public void describeTo(Description description) {
+        description.appendText(String.format("NodeEvent with description '%s'", expected));
+    }
+
+    @Override
+    public void describeMismatch(Object item, Description description) {
+        NodeEvent other = (NodeEvent)item;
+        description.appendText(String.format("got description '%s'", other.getDescription()));
+    }
+
+    @Factory
+    public static NodeEventWithDescription nodeEventWithDescription(String description) {
+        return new NodeEventWithDescription(description);
+    }
+}