Accord: Add Rebootstrap and unsafe Bootstrap

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
-	url = https://github.com/apache/cassandra-accord.git
-	branch = trunk
+	url = https://github.com/belliottsmith/cassandra-accord.git
+	branch = rebootstrap-sq

src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java

@@ -47,6 +47,7 @@
 import accord.coordinate.Coordinations;
 import accord.coordinate.PrepareRecovery;
 import accord.coordinate.tracking.AbstractTracker;
+import accord.primitives.RoutingKeys;
 import accord.utils.SortedListMap;
 import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.db.EmptyIterators;
@@ -74,8 +75,6 @@
 import accord.local.CommandStores.LatentStoreSelector;
 import accord.local.Commands;
 import accord.local.DurableBefore;
-import accord.local.LoadKeys;
-import accord.local.LoadKeysFor;
 import accord.local.MaxConflicts;
 import accord.local.Node;
 import accord.local.PreLoadContext;
@@ -143,8 +142,9 @@
 import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
 import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED;
 import static accord.local.RedundantStatus.Property.LOCALLY_WITNESSED;
+import static accord.local.RedundantStatus.Property.LOG_UNAVAILABLE;
 import static accord.local.RedundantStatus.Property.QUORUM_APPLIED;
-import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP;
+import static accord.local.RedundantStatus.Property.UNREADY;
 import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.lang.String.format;
@@ -365,9 +365,8 @@ public Partition getPartition(DecoratedKey partitionKey)
             TokenKey key = TokenKey.parse(keyStr, DatabaseDescriptor.getPartitioner());
 
             List<Entry> cfks = new CopyOnWriteArrayList<>();
-            PreLoadContext context = PreLoadContext.contextFor(key, LoadKeys.SYNC, LoadKeysFor.READ_WRITE, "commands_for_key table query");
             CommandStores commandStores = AccordService.instance().node().commandStores();
-            AccordService.getBlocking(commandStores.forEach(context, key, Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> {
+            AccordService.getBlocking(commandStores.forEach("commands_for_key table query", RoutingKeys.of(key), Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> {
                 SafeCommandsForKey safeCfk = safeStore.get(key);
                 CommandsForKey cfk = safeCfk.current();
                 if (cfk == null)
@@ -475,9 +474,8 @@ public Partition getPartition(DecoratedKey partitionKey)
             TokenKey key = TokenKey.parse(keyStr, DatabaseDescriptor.getPartitioner());
 
             List<Entry> cfks = new CopyOnWriteArrayList<>();
-            PreLoadContext context = PreLoadContext.contextFor(key, LoadKeys.SYNC, LoadKeysFor.READ_WRITE, "commands_for_key_unmanaged table query");
             CommandStores commandStores = AccordService.instance().node().commandStores();
-            AccordService.getBlocking(commandStores.forEach(context, key, Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> {
+            AccordService.getBlocking(commandStores.forEach("commands_for_key_unmanaged table query", RoutingKeys.of(key), Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> {
                 SafeCommandsForKey safeCfk = safeStore.get(key);
                 CommandsForKey cfk = safeCfk.current();
                 if (cfk == null)
@@ -888,8 +886,9 @@ private RedundantBeforeTable()
                         "  locally_redundant 'TxnIdUtf8Type',\n" +
                         "  locally_synced 'TxnIdUtf8Type',\n" +
                         "  locally_witnessed 'TxnIdUtf8Type',\n" +
-                        "  pre_bootstrap 'TxnIdUtf8Type',\n" +
-                        "  stale_until_at_least 'TxnIdUtf8Type',\n" +
+                        "  log_unavailable 'TxnIdUtf8Type',\n" +
+                        "  unready 'TxnIdUtf8Type',\n" +
+                        "  stale_until 'TxnIdUtf8Type',\n" +
                         "  PRIMARY KEY (keyspace_name, table_name, table_id, command_store_id, token_start)" +
                         ')', UTF8Type.instance));
         }
@@ -923,8 +922,9 @@ public DataSet data()
                           .column("locally_redundant", entry.maxBound(LOCALLY_REDUNDANT).toString())
                           .column("locally_synced", entry.maxBound(LOCALLY_SYNCED).toString())
                           .column("locally_witnessed", entry.maxBound(LOCALLY_WITNESSED).toString())
-                          .column("pre_bootstrap", entry.maxBound(PRE_BOOTSTRAP).toString())
-                          .column("stale_until_at_least", entry.staleUntilAtLeast != null ? entry.staleUntilAtLeast.toString() : null);
+                          .column("log_unavailable", entry.maxBound(LOG_UNAVAILABLE).toString())
+                          .column("unready", entry.maxBound(UNREADY).toString())
+                          .column("stale_until", entry.staleUntilAtLeast != null ? entry.staleUntilAtLeast.toString() : null);
                         return ds;
                     },
                     dataSet,
@@ -1188,7 +1188,7 @@ public Partition getPartition(DecoratedKey partitionKey)
             TxnId txnId = TxnId.parse(txnIdStr);
 
             List<Entry> commands = new CopyOnWriteArrayList<>();
-            AccordService.instance().node().commandStores().forEachCommandStore(store -> {
+            AccordService.instance().node().commandStores().forAllUnsafe(store -> {
                 Command command = ((AccordCommandStore)store).loadCommand(txnId);
                 if (command != null)
                     commands.add(new Entry(store.id(), command));
@@ -1293,7 +1293,7 @@ public Partition getPartition(DecoratedKey partitionKey)
             TxnId txnId = TxnId.parse(txnIdStr);
 
             List<Entry> entries = new ArrayList<>();
-            AccordService.instance().node().commandStores().forEachCommandStore(store -> {
+            AccordService.instance().node().commandStores().forAllUnsafe(store -> {
                 for (AccordJournal.DebugEntry e : ((AccordCommandStore)store).debugCommand(txnId))
                     entries.add(new Entry(store.id(), e.segment, e.position, e.builder));
             });

src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java

@@ -175,7 +175,7 @@ private synchronized void refreshHistograms()
 
         int nowSeconds = (int) (Clock.Global.currentTimeMillis() / 1000);
         SnapshotBuilder builder = new SnapshotBuilder();
-        service.node().commandStores().forEachCommandStore(commandStore -> {
+        service.node().commandStores().forAllUnsafe(commandStore -> {
             DefaultProgressLog.ImmutableView view = ((DefaultProgressLog)commandStore.unsafeProgressLog()).immutableView();
             builder.progressLogActive += view.activeCount();
             builder.progressLogSize.increment(view.size());

src/java/org/apache/cassandra/service/ActiveRepairService.java

@@ -582,8 +582,7 @@ else if (range.intersects(toRepair))
             {
                 throw new IllegalArgumentException(String.format("Requested range %s intersects a local range (%s) " +
                                                                  "but is not fully contained in one; this would lead to " +
-                                                                 "imprecise repair. keyspace: %s", toRepair.toString(),
-                                                                 range.toString(), keyspaceName));
+                                                                 "imprecise repair. keyspace: %s", toRepair, range, keyspaceName));
             }
         }
         if (rangeSuperSet == null || !replicaSets.containsKey(rangeSuperSet))

src/java/org/apache/cassandra/service/Rebuild.java

@@ -33,6 +33,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import accord.api.ConfigurationService.EpochReady;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.dht.Range;
@@ -160,7 +161,7 @@ else if (tokens == null)
 
             StreamResultFuture streamResult = streamer.fetchAsync();
 
-            Future<?> accordReady = AccordService.instance().epochReadyFor(metadata);
+            Future<?> accordReady = AccordService.instance().epochReadyFor(metadata, EpochReady::reads);
             Future<?> ready = FutureCombiner.allOf(streamResult, accordReady);
 
             // wait for result

src/java/org/apache/cassandra/service/StorageService.java

@@ -71,9 +71,6 @@
 import com.google.common.collect.Ordering;
 import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.Uninterruptibles;
-
-import org.apache.cassandra.io.sstable.format.SSTableReader;
-import org.apache.cassandra.repair.autorepair.AutoRepair;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -134,6 +131,7 @@
 import org.apache.cassandra.io.sstable.IScrubber;
 import org.apache.cassandra.io.sstable.IVerifier;
 import org.apache.cassandra.io.sstable.SSTableLoader;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.format.Version;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileUtils;
@@ -158,7 +156,9 @@
 import org.apache.cassandra.metrics.StorageMetrics;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.repair.RepairCoordinator;
+import org.apache.cassandra.repair.RepairParallelism;
 import org.apache.cassandra.repair.SharedContext;
+import org.apache.cassandra.repair.autorepair.AutoRepair;
 import org.apache.cassandra.repair.messages.RepairOption;
 import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
 import org.apache.cassandra.schema.KeyspaceMetadata;
@@ -183,6 +183,7 @@
 import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupLocalCoordinator;
 import org.apache.cassandra.service.paxos.cleanup.PaxosRepairState;
 import org.apache.cassandra.service.snapshot.SnapshotManager;
+import org.apache.cassandra.streaming.PreviewKind;
 import org.apache.cassandra.streaming.StreamManager;
 import org.apache.cassandra.streaming.StreamResultFuture;
 import org.apache.cassandra.streaming.StreamState;
@@ -206,9 +207,9 @@
 import org.apache.cassandra.tcm.sequences.BootstrapAndReplace;
 import org.apache.cassandra.tcm.sequences.InProgressSequences;
 import org.apache.cassandra.tcm.sequences.SingleNodeSequences;
+import org.apache.cassandra.tcm.transformations.AlterTopology;
 import org.apache.cassandra.tcm.transformations.Assassinate;
 import org.apache.cassandra.tcm.transformations.CancelInProgressSequence;
-import org.apache.cassandra.tcm.transformations.AlterTopology;
 import org.apache.cassandra.tcm.transformations.Register;
 import org.apache.cassandra.tcm.transformations.Startup;
 import org.apache.cassandra.tcm.transformations.Unregister;
@@ -3145,6 +3146,29 @@ private FutureTask<Object> createRepairTask(final int cmd, final String keyspace
         return new FutureTask<>(task);
     }
 
+    public RepairCoordinator repairAccordKeyspace(String keyspace, Collection<Range<Token>> ranges)
+    {
+        int cmd = nextRepairCommand.incrementAndGet();
+        RepairOption options = new RepairOption(RepairParallelism.PARALLEL, // parallelism
+                                                false,                       // primaryRange
+                                                false,                      // incremental
+                                                false,                      // trace
+                                                5,                          // jobThreads
+                                                ranges,                     // ranges
+                                                true,                       // pullRepair
+                                                true,                       // forceRepair
+                                                PreviewKind.NONE,           // previewKind
+                                                false,                      // optimiseStreams
+                                                true,                       // ignoreUnreplicatedKeyspaces
+                                                true,                       // repairData
+                                                false,                      // repairPaxos
+                                                true,                       // dontPurgeTombstones
+                                                false                       // repairAccord
+        );
+
+        return new RepairCoordinator(this, cmd, options, keyspace);
+    }
+
     private void tryRepairPaxosForTopologyChange(String reason)
     {
         try

src/java/org/apache/cassandra/service/accord/AccordCommandStores.java

@@ -37,9 +37,9 @@
 import org.apache.cassandra.cache.CacheSize;
 import org.apache.cassandra.config.AccordSpec.QueueShardModel;
 import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.service.accord.AccordExecutor.AccordExecutorFactory;
-import org.apache.cassandra.service.accord.api.TokenKey;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.NodeState;
 
 import static org.apache.cassandra.config.AccordSpec.QueueShardModel.THREAD_PER_SHARD;
 import static org.apache.cassandra.config.DatabaseDescriptor.getAccordQueueShardCount;
@@ -112,33 +112,13 @@ static Factory factory()
         };
     }
 
-    @Override
-    protected boolean shouldBootstrap(Node node, Topology previous, Topology updated, Range range)
-    {
-        if (!super.shouldBootstrap(node, previous, updated, range))
-            return false;
-        // we see new ranges when a new keyspace is added, so avoid bootstrap in these cases
-        return contains(previous, ((TokenKey)  range.start()).table());
-    }
-
     @Override
     public SequentialAsyncExecutor someSequentialExecutor()
     {
         int idx = ((int) Thread.currentThread().getId()) & mask;
         return executors[idx].newSequentialExecutor();
     }
 
-    private static boolean contains(Topology previous, TableId searchTable)
-    {
-        for (Range range : previous.ranges())
-        {
-            TableId table = ((TokenKey)  range.start()).table();
-            if (table.equals(searchTable))
-                return true;
-        }
-        return false;
-    }
-
     public synchronized void setCapacity(long bytes)
     {
         cacheSize = bytes;

src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java

@@ -113,7 +113,7 @@ AsyncResult<Void> acknowledged()
 
         @Nullable AsyncResult<Void> reads()
         {
-            return reads;
+            return ready == null ? null : ready.reads;
         }
 
         AsyncResult.Settable<Void> localSyncNotified()
@@ -449,7 +449,7 @@ public void reportTopology(Topology topology, boolean isLoad, boolean startSync)
     }
 
     @Override
-    protected void localSyncComplete(Topology topology, boolean startSync)
+    protected void onReadyToCoordinate(Topology topology, boolean startSync)
     {
         long epoch = topology.epoch();
         EpochState epochState = getOrCreateEpochState(epoch);

src/java/org/apache/cassandra/service/accord/AccordDataStore.java

@@ -30,6 +30,7 @@
 import accord.primitives.Range;
 import accord.primitives.Ranges;
 import accord.primitives.SyncPoint;
+import accord.utils.UnhandledEnum;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.memtable.Memtable;
 import org.apache.cassandra.schema.Schema;
@@ -40,14 +41,6 @@ public class AccordDataStore implements DataStore
     private static final Logger logger = LoggerFactory.getLogger(AccordDataStore.class);
     enum FlushListenerKey { KEY }
 
-    @Override
-    public FetchResult fetch(Node node, SafeCommandStore safeStore, Ranges ranges, SyncPoint syncPoint, FetchRanges callback)
-    {
-        AccordFetchCoordinator coordinator = new AccordFetchCoordinator(node, ranges, syncPoint, callback, safeStore.commandStore());
-        coordinator.start();
-        return coordinator.result();
-    }
-
     /**
      * Ensures data for the intersecting ranges is flushed to sstable before calling back with reportOnSuccess.
      * This is used to gate journal cleanup, since we skip the CommitLog for applying to the data table.
@@ -95,4 +88,23 @@ public void ensureDurable(CommandStore commandStore, Ranges ranges, RedundantBef
             prev = cfs;
         }
     }
+
+    @Override
+    public FetchResult fetch(Node node, SafeCommandStore safeStore, Ranges ranges, SyncPoint syncPoint, FetchRanges callback, FetchKind kind)
+    {
+        switch (kind)
+        {
+            default: throw new UnhandledEnum(kind);
+            case Image:
+            {
+                AccordFetchCoordinator coordinator = new AccordFetchCoordinator(node, ranges, syncPoint, callback, safeStore.commandStore());
+                coordinator.start();
+                return coordinator.result();
+            }
+            case Sync:
+            {
+                throw new UnsupportedOperationException();
+            }
+        }
+    }
 }

src/java/org/apache/cassandra/service/accord/AccordJournal.java

@@ -585,7 +585,7 @@ public void forEach(Consumer<JournalKey> consumer)
 
     @SuppressWarnings("unchecked")
     @Override
-    public void replay(CommandStores commandStores)
+    public boolean replay(CommandStores commandStores)
     {
         // TODO (expected): make the parallelisms configurable
         // Replay is performed in parallel, where at most X commands can be in flight, accross at most Y commands stores.
@@ -716,6 +716,7 @@ public void close()
 
                 ++cur;
             }
+            return true;
         }
         catch (Throwable t)
         {