Introduce "hardRemoved" concept for marking nodes permanently offline, to permit recovery from availability loss

Also fix:
 - Paxos should not update minHlc if AccordService is not setup
 - Remove EndpointMapper methods that require non-null, to ensure call-sites handle null case explicitly
 - AccordRepair should specify the Node Ids that must participate, but these should not affect the durability requirements for GC
 - Avoid erroneously marking UniversalOrInvalidated when only known to quorum
 - AccordSyncPropagator should not consult FailureDetector as it may throw UnknownEndpointException
 - Refuse ReplaceSameAddress if accord enabled
 - Reset Accord topologies on restart if no local command stores to ensure we can catch up (as intervening epochs may have otherwise been GC'd)
 - MaybeRecover should not abort if found durable if either: 1) prevProgressToken already witnessed durable; or 2) Home state has requested we recover anyway
 - Invariant in MaybeRecover is too strong now we permit recovering when known to be stable
 - ReadData must respond with InsufficientAndWaiting if not Stable to ensure Durability is inferred correctly by ExecuteTxn
 - Avoid StackOverflowException in NotifyWaitingOn
 - PreAccept should honour the REJECTED flag of any member of the latest topology if the vote would be relied upon for any prior epoch's quorum
 - Resume Accord bootstrap on restart
 - Fix burn test non-determinism
 - Reset Accord topologies on restart if no local command stores to ensure we can catch up (as intervening epochs may have otherwise been GC'd)
 - Exclude stale or removed ids from sync points
 - Terminate a failed DurabilityRequest if a required participant is now removed/hardRemoved
Also improve:
 - Introduce accord_debug_keyspace.{listeners_txn, listeners_local}
 - Filter and record faulty in AbstractCoordination
 - If we are the home shard, and our home progress status is done, we should not wait for the home shard to advance the waiting progress state
 - Introduce CommandStore.tryExecuteListening to try to execute all waiting transactions and their dependencies
 - ExecuteTxn on recovery should not wait, to avoid consuming progress log concurrency slots for transactions that cannot execute
 - HomeState should update its phase based on the command status to handle operator request to retry all states
 - Separate home/wait queue in DefaultProgressLog
 - Additional tracing
 - Do not call trySendMore from prerecordFailure
 - Force full recovery for sync points if we have lost quorum

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20921

Author: belliottsmith

modules/accord

@@ -143,11 +143,12 @@ public enum QueueSubmissionModel
     public String slow_syncpoint_preaccept = "10s";
     public String slow_txn_preaccept = "30ms <= p50*2 <= 100ms";
     public String slow_read = "30ms <= p50*2 <= 100ms";
-    public StringRetryStrategy retry_syncpoint = new StringRetryStrategy("10s*attempts <= 600s");
-    public StringRetryStrategy retry_durability = new StringRetryStrategy("10s*attempts <= 600s");
-    public StringRetryStrategy retry_bootstrap = new StringRetryStrategy("10s*attempts <= 600s");
-    public StringRetryStrategy retry_fetch_min_epoch = new StringRetryStrategy("200ms...1s*attempts <= 1s,retries=3");
-    public StringRetryStrategy retry_fetch_topology = new StringRetryStrategy("200ms...1s*attempts <= 1s,retries=100");
+    public StringRetryStrategy retry_syncpoint = new StringRetryStrategy("10s*attempt <= 600s");
+    public StringRetryStrategy retry_durability = new StringRetryStrategy("10s*attempt <= 600s");
+    public StringRetryStrategy retry_bootstrap = new StringRetryStrategy("10s*attempt <= 600s");
+    public StringRetryStrategy retry_join_bootstrap = new StringRetryStrategy("30s*attempt,attempts=5");
+    public StringRetryStrategy retry_fetch_min_epoch = new StringRetryStrategy("200ms...1s*attempt <= 1s,retries=3");
+    public StringRetryStrategy retry_fetch_topology = new StringRetryStrategy("200ms...1s*attempt <= 1s,retries=100");
     public StringRetryStrategy retry_journal_index_ready = new StringRetryStrategy("100ms");
 
     public volatile DurationSpec.IntSecondsBound fast_path_update_delay = null;

src/java/org/apache/cassandra/config/AccordSpec.java

@@ -26,6 +26,6 @@ public class AccordDebugRemoteKeyspace extends RemoteToLocalVirtualKeyspace
 
     public AccordDebugRemoteKeyspace(String name, VirtualKeyspace wrap)
     {
-        super(name, wrap);
+        super(name, wrap, vt -> !vt.name().equals(AccordDebugKeyspace.NODE_OPS));
     }
 }

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

@@ -43,7 +43,7 @@ public void register(VirtualKeyspace keyspace)
         VirtualKeyspace previous = virtualKeyspaces.put(keyspace.name(), keyspace);
         // some tests choose to replace the keyspace, if so make sure to cleanup tables as well
         if (previous != null)
-            previous.tables().forEach(t -> virtualTables.remove(t));
+            previous.tables().forEach(t -> virtualTables.remove(t.metadata().id));
         keyspace.tables().forEach(t -> virtualTables.put(t.metadata().id, t));
     }
 

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

@@ -63,7 +63,7 @@
  * need to sometimes return a port and sometimes not.
  *
  */
-public final class InetAddressAndPort extends InetSocketAddress implements Comparable<InetAddressAndPort>, Serializable
+public final class InetAddressAndPort extends InetSocketAddress implements Comparable<InetAddressAndPort>, Serializable, Endpoint
 {
     private static final long serialVersionUID = 0;
     private static final Logger logger = LoggerFactory.getLogger(InetAddressAndPort.class);
@@ -382,6 +382,12 @@ static int getDefaultPort()
         return defaultPort;
     }
 
+    @Override
+    public InetAddressAndPort endpoint()
+    {
+        return this;
+    }
+
     public static final class MetadataSerializer implements org.apache.cassandra.tcm.serialization.MetadataSerializer<InetAddressAndPort>
     {
         public static final MetadataSerializer serializer = new MetadataSerializer();

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

@@ -192,7 +192,7 @@ protected void runRepair()
                         requireAllEndpoints = true;
                 }
                 logger.info("{} {}.{} starting accord repair, require all endpoints {}", session.previewKind.logPrefix(session.getId()), desc.keyspace, desc.columnFamily, requireAllEndpoints);
-                AccordRepair repair = new AccordRepair(ctx, cfs, desc.sessionId, desc.keyspace, desc.ranges, requireAllEndpoints);
+                AccordRepair repair = new AccordRepair(ctx, cfs, desc.sessionId, desc.keyspace, desc.ranges, requireAllEndpoints, allEndpoints);
                 return repair.repair(taskExecutor).flatMap(accordRepairResult -> {
                     // Propagate the HLC discovered during Accord repair to Paxos so Paxos doesn't use ballots < Accord has already used
                     if (accordRepairResult.maxHlc != IAccordService.NO_HLC)

src/java/org/apache/cassandra/locator/InetAddressAndPort.java

@@ -276,6 +276,13 @@ public static RetryStrategy parse(String spec, LatencySourceFactory latencies, W
                     default: throw new IllegalArgumentException("Invalid modifier specification: unrecognised property '" + key + '\'');
                     case "retries":
                         retries = Integer.parseInt(value);
+                        if (retries < 0)
+                            throw new IllegalArgumentException("retries must be non-negative (retries=" + value + " supplied)");
+                        break;
+                    case "attempts":
+                        retries = Integer.parseInt(value);
+                        if (retries < 0)
+                            throw new IllegalArgumentException("Must permit at least one attempt (attempts=" + value + " supplied)");
                         break;
                     case "rnd":
                         if (randomizer != null)

src/java/org/apache/cassandra/repair/RepairJob.java

@@ -85,7 +85,7 @@ public class TimeoutStrategy implements WaitStrategy
     static final Pattern WAIT = Pattern.compile(
                 "\\s*(?<const>0|[0-9]+[mu]?s)" +
                 "|\\s*((p(?<perc>[0-9]+)(\\((?<rw>r|w|rw|wr)\\))?)?|(?<constbase>0|[0-9]+[mu]?s))" +
-                    "\\s*(([*]\\s*(?<mod>[0-9.]+))?\\s*(?<modkind>[*^]\\s*attempts)?)?\\s*");
+                    "\\s*(([*]\\s*(?<mod>[0-9.]+))?\\s*(?<modkind>[*^]\\s*attempts?)?)?\\s*");
     static final Pattern TIME = Pattern.compile(
                 "0|[0-9]+[mu]?s");
 

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

@@ -474,7 +474,7 @@ public AccordCommandStoreReplayer replayer()
     @Override
     protected void ensureDurable(Ranges ranges, RedundantBefore onCommandStoreDurable)
     {
-        if (!CommandsForKey.reportLinearizabilityViolations())
+        if (node().isReplaying())
             return;
 
         long reportId = nextDurabilityLoggingId.incrementAndGet();