Fix topology loading

  - in burn test, use validateShardStateForTesting to ensure that latest topology update matches the shard state, which simulates book
  - add an initializeTopologyUnsafe for an implementation to use to _create_ shards during boot so that shards could be initialized without replaying all preceding topologies

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20294

Author: ifesdjeen

accord-core/src/main/java/accord/api/Journal.java

@@ -47,7 +47,7 @@ public interface Journal
     // TODO (required): use OnDone instead of Runnable
     void saveCommand(int store, CommandUpdate value, Runnable onFlush);
 
-    Iterator<TopologyUpdate> replayTopologies(); // reverse iterator
+    Iterator<TopologyUpdate> replayTopologies();
     void saveTopology(TopologyUpdate topologyUpdate, Runnable onFlush);
 
     void purge(CommandStores commandStores);

accord-core/src/main/java/accord/local/CommandStores.java

@@ -20,7 +20,6 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -133,7 +132,7 @@ protected static class ShardHolder
             this.store = store;
         }
 
-        private ShardHolder(CommandStore store, RangesForEpoch ranges)
+        public ShardHolder(CommandStore store, RangesForEpoch ranges)
         {
             this.store = store;
             this.ranges = ranges;
@@ -144,7 +143,7 @@ public ShardHolder withStoreUnsafe(CommandStore store)
             return new ShardHolder(store, ranges);
         }
 
-        RangesForEpoch ranges()
+        public RangesForEpoch ranges()
         {
             return ranges;
         }
@@ -381,43 +380,6 @@ public Ranges removed(long presentIn, long removedByInclusive)
         }
     }
 
-    // This method should only be used on node startup.
-    // "Unsafe" because it relies on user to synchronise and sequence the call properly.
-    public void restoreShardStateUnsafe(Consumer<Topology> reportTopology)
-    {
-        Iterator<Journal.TopologyUpdate> iter = journal.replayTopologies();
-        // First boot
-        if (!iter.hasNext())
-            return;
-
-        Journal.TopologyUpdate lastUpdate = null;
-        while (iter.hasNext())
-        {
-            Journal.TopologyUpdate update = iter.next();
-            reportTopology.accept(update.global);
-            if (lastUpdate == null || update.global.epoch() > lastUpdate.global.epoch())
-                lastUpdate = update;
-        }
-
-        ShardHolder[] shards = new ShardHolder[lastUpdate.commandStores.size()];
-        int i = 0;
-        for (Map.Entry<Integer, RangesForEpoch> e : lastUpdate.commandStores.entrySet())
-        {
-            RangesForEpoch ranges = e.getValue();
-            CommandStore commandStore = null;
-            for (ShardHolder shard : current.shards)
-            {
-                if (shard.ranges.equals(ranges))
-                    commandStore = shard.store;
-            }
-            Invariants.nonNull(commandStore, "Command store should have been reloaded").restore();
-            ShardHolder shard = new ShardHolder(commandStore, e.getValue());
-            shards[i++] = shard;
-        }
-
-        loadSnapshot(new Snapshot(shards, lastUpdate.local, lastUpdate.global));
-    }
-
     protected void loadSnapshot(Snapshot toLoad)
     {
         current = toLoad;
@@ -428,7 +390,7 @@ protected static class Snapshot extends Journal.TopologyUpdate
         public final ShardHolder[] shards;
         public final Int2ObjectHashMap<CommandStore> byId;
 
-        Snapshot(ShardHolder[] shards, Topology local, Topology global)
+        public Snapshot(ShardHolder[] shards, Topology local, Topology global)
         {
             super(asMap(shards), local, global);
             this.shards = shards;
@@ -771,6 +733,20 @@ protected <O> AsyncChain<O> mapReduce(PreLoadContext context, MapReduce<? super
         return chain == null ? AsyncChains.success(null) : chain;
     }
 
+    /**
+     * Initialize topology from snapshot on boot.
+     */
+    public synchronized void initializeTopologyUnsafe(Journal.TopologyUpdate update)
+    {
+        Invariants.require(current.global.epoch() == 0);
+        ShardHolder[] shards = new ShardHolder[update.commandStores.size()];
+        int i = 0;
+        for (Map.Entry<Integer, RangesForEpoch> e : update.commandStores.entrySet())
+            shards[i++] = new ShardHolder(supplier.create(e.getKey(), new EpochUpdateHolder()), e.getValue());
+
+        loadSnapshot(new Snapshot(shards, update.local, update.global));
+    }
+
     public synchronized Supplier<EpochReady> updateTopology(Node node, Topology newTopology, boolean startSync)
     {
         TopologyUpdate update = updateTopology(node, current, newTopology, startSync);

accord-core/src/test/java/accord/impl/basic/Cluster.java

@@ -764,15 +764,23 @@ public static Map<MessageType, Stats> run(Id[] nodes, int[] prefixes, MessageLis
                 Int2ObjectHashMap<NavigableMap<TxnId, Command>> beforeStores = copyCommands(stores.all());
 
                 for (CommandStore store : stores.all())
-                {
                     ((InMemoryCommandStore) store).clearForTesting();
-                }
-                // Re-create all command stores
-                nodeMap.get(id).commandStores().restoreShardStateUnsafe(t -> {});
-                stores = nodeMap.get(id).commandStores();
 
                 // Replay journal
                 Journal journal = journalMap.get(id);
+                Iterator<Journal.TopologyUpdate> iter = journal.replayTopologies();
+                Journal.TopologyUpdate lastUpdate = null;
+                while (iter.hasNext())
+                {
+                    Journal.TopologyUpdate update = iter.next();
+                    Invariants.require(lastUpdate == null || update.global.epoch() > lastUpdate.global.epoch());
+                    lastUpdate = update;
+                }
+
+                if (lastUpdate != null)
+                    ((DelayedCommandStores) nodeMap.get(id).commandStores()).validateShardStateForTesting(lastUpdate);
+
+                stores = nodeMap.get(id).commandStores();
                 journal.replay(stores);
 
                 // Re-enable safety checks

accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java

@@ -19,6 +19,7 @@
 package accord.impl.basic;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -93,6 +94,30 @@ protected boolean shouldBootstrap(Node node, Topology previous, Topology updated
         return contains(previous, prefix);
     }
 
+    public void validateShardStateForTesting(Journal.TopologyUpdate lastUpdate)
+    {
+        ShardHolder[] shards = new ShardHolder[lastUpdate.commandStores.size()];
+        int i = 0;
+        for (Map.Entry<Integer, RangesForEpoch> e : lastUpdate.commandStores.entrySet())
+        {
+            Snapshot current = current();
+            RangesForEpoch ranges = e.getValue();
+            CommandStore commandStore = null;
+            for (ShardHolder shard : current.shards)
+            {
+                if (shard.ranges().equals(ranges))
+                    commandStore = shard.store;
+            }
+            Invariants.nonNull(commandStore, "Each set of ranges should have a corresponding command store, but %d did not:(%s)",
+                               ranges, Arrays.toString(shards))
+                      .restore();
+            ShardHolder shard = new ShardHolder(commandStore, e.getValue());
+            shards[i++] = shard;
+        }
+
+        loadSnapshot(new Snapshot(shards, lastUpdate.local, lastUpdate.global));
+    }
+
     protected void loadSnapshot(Snapshot nextSnapshot)
     {
         Snapshot current = current();