Refactor RedundantStatus to encode vector of states that can be merged independently

Also fix:
 - Truncate command on first access, without participants
 - Use Ballot.ZERO when invoking CFK.insertOutOfRange where appropriate
 - Don't supply a command's own route to ProgressLog.waiting to ensure new keys are incorporated
 - Ensure progress in CommandsForKey by setting vestigial commands to ERASED
 - Add any missing owned keys to StoreParticipants.route to ensure fetch can make progress
 - Recovery must wait for earlier not-accepted transactions if either has the privileged coordinator optimisation
 - Inclusive SyncPoint used incorrect topologies for propose phase
 - Barrier must not register local listener without up-to-date topology information
 - Stop home shard truncating a TxnId to vestigial rather than Invalidated so other shards can make progress
Also improve:
 - Validate commands are constructed with non-empty participants
 - Remove some unnecessary synchronized keywords
 - Clear ok messages on PreAccept and Accept to free up memory
 - Introduce TxnId.Cardinality flag so we can optimise single key queries
 - Update CommandsForKey serialization to better handle larger flag space
 - Configurable which Txn.Kind can result in a CommandStore being marked stale
 - Process DefaultProgressLog queue synchronously when relevant state is resident in memory
 - Remove defunct CollectMaxApplied version of ListStore bootstrap
 - Standardise linearizability violation reporting
 - Improve CommandStore.execute method naming to reduce chance of misuse
 - Relax Invariant on recovery that we cannot both Accept and Reject, as we could be a stale recovery
 - Prune and address some comments

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

Author: belliottsmith

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

@@ -20,11 +20,14 @@
 
 import java.util.concurrent.TimeUnit;
 
+import javax.annotation.Nullable;
+
 import accord.api.ProgressLog.BlockedUntil;
 import accord.local.Command;
 import accord.local.Node;
 import accord.local.SafeCommandStore;
 import accord.messages.ReplyContext;
+import accord.primitives.Participants;
 import accord.primitives.Ranges;
 import accord.primitives.Routable;
 import accord.primitives.Status.Phase;
@@ -153,5 +156,5 @@ default EventsListener metricsEventsListener()
     long localExpiresAt(TxnId txnId, Phase phase, TimeUnit unit);
     long expiresAt(ReplyContext replyContext, TimeUnit unit);
 
-    default void onViolation(String message) { throw illegalState(message); }
+    default void onViolation(String message, Participants<?> participants, @Nullable TxnId notWitnessed, @Nullable Timestamp notWitnessedExecuteAt, @Nullable TxnId by, @Nullable Timestamp byEexecuteAt) { throw illegalState(message); }
 }

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

@@ -21,11 +21,13 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import accord.primitives.Txn;
 import accord.primitives.TxnId;
 import accord.primitives.TxnId.FastPath;
 import accord.primitives.TxnId.FastPaths;
 import accord.primitives.TxnId.MediumPath;
 import accord.utils.Invariants;
+import accord.utils.TinyEnumSet;
 
 import static accord.api.ProtocolModifiers.QuorumEpochIntersections.ChaseFixedPoint.Chase;
 import static accord.api.ProtocolModifiers.QuorumEpochIntersections.ChaseFixedPoint.DoNotChase;
@@ -162,7 +164,6 @@ static class Spec
 
         static
         {
-            // TODO (expected): configurable
             Spec spec = new Spec(false, false, false, false);
             txnInstability = spec.txnInstability;
             txnDiscardPreAcceptDeps = spec.txnDiscardPreAcceptDeps;
@@ -187,7 +188,7 @@ public static class Toggles
         public static void setPermittedFastPaths(FastPaths newPermittedFastPaths) { permittedFastPaths = newPermittedFastPaths; }
         public static FastPath ensurePermitted(FastPath path) { return path.toPermitted(permittedFastPaths); }
 
-        private static MediumPath defaultMediumPath = MediumPath.TRACK_STABLE;
+        private static MediumPath defaultMediumPath = MediumPath.TrackStable;
         public static MediumPath defaultMediumPath() { return defaultMediumPath; }
         public static void setDefaultMediumPath(MediumPath newDefaultMediumPath) { defaultMediumPath = newDefaultMediumPath; }
 
@@ -203,6 +204,15 @@ public static class Toggles
         public static boolean requiresUniqueHlcs() { return requiresUniqueHlcs; }
         public static void setRequiresUniqueHlcs(boolean newRequiresUniqueHlcs) { requiresUniqueHlcs = newRequiresUniqueHlcs; }
 
+        private static int markStaleIfCannotExecute = TinyEnumSet.encode(Txn.Kind.Write);
+        public static boolean markStaleIfCannotExecute(Txn.Kind kind) { return TinyEnumSet.contains(markStaleIfCannotExecute, kind); }
+        public static void setMarkStaleIfCannotExecute(Txn.Kind ... kinds)
+        {
+            int newMarkStaleIfCannotExecute = TinyEnumSet.encode(kinds);
+            Invariants.require(TinyEnumSet.contains(newMarkStaleIfCannotExecute, Txn.Kind.Write));
+            markStaleIfCannotExecute = newMarkStaleIfCannotExecute;
+        }
+
         public enum DependencyElision { OFF, ON, IF_DURABLE }
         private static DependencyElision dependencyElision = IF_DURABLE;
         public static DependencyElision dependencyElision() { return dependencyElision; }

accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java

@@ -74,30 +74,29 @@ void onSlowResponseInternal(Id from) {}
     abstract long executeAtEpoch();
 
     @Override
-    public synchronized final void onFailure(Id from, Throwable failure)
+    public final void onFailure(Id from, Throwable failure)
     {
         if (!isDone)
             onFailureInternal(from, failure);
     }
 
     @Override
-    public final synchronized boolean onCallbackFailure(Id from, Throwable failure)
+    public final boolean onCallbackFailure(Id from, Throwable failure)
     {
         if (isDone) return false;
         isDone = true;
         return tryFailure(failure);
     }
 
     @Override
-    // TODO (expected): shouldn't need synchronized
-    public final synchronized void onSuccess(Id from, R reply)
+    public final void onSuccess(Id from, R reply)
     {
         if (!isDone)
             onSuccessInternal(from, reply);
     }
 
     @Override
-    public final synchronized void onSlowResponse(Id from)
+    public final void onSlowResponse(Id from)
     {
         if (!isDone)
             onSlowResponseInternal(from);

accord-core/src/main/java/accord/coordinate/Barrier.java

@@ -57,6 +57,7 @@
 import static accord.local.CommandSummaries.ComputeIsDep.IGNORE;
 import static accord.local.CommandSummaries.TestStartedAt.STARTED_AFTER;
 import static accord.primitives.Txn.Kind.AnyGloballyVisible;
+import static accord.primitives.TxnId.Cardinality.cardinality;
 import static accord.utils.Invariants.illegalState;
 
 /**
@@ -149,7 +150,7 @@ public static Barrier barrier(Node node, Seekables<?, ?> keysOrRanges, FullRoute
     private static BiFunction<Node, FullRoute<?>, AsyncSyncPoint> wrap(TriFunction<Node, TxnId, FullRoute<?>, AsyncResult<? extends SyncPoint<?>>> syncPointFactory)
     {
         return (node, route) -> {
-            TxnId txnId = node.nextTxnId(Txn.Kind.SyncPoint, route.domain());
+            TxnId txnId = node.nextTxnId(Txn.Kind.SyncPoint, route.domain(), cardinality(route));
             return new AsyncSyncPoint(txnId, syncPointFactory.apply(node, txnId, route));
         };
     }
@@ -186,11 +187,10 @@ private void createSyncPoint()
         coordinateSyncPoint = async.async;
         if (barrierType.async)
         {
-            Invariants.require(barrierType.async);
             TxnId txnId = async.txnId;
             long epoch = txnId.epoch();
             RoutingKey homeKey = route.homeKey();
-            node.commandStores().ifLocal(txnId, homeKey, epoch, epoch, safeStore -> register(safeStore, txnId, homeKey))
+            node.withEpoch(epoch, () -> node.commandStores().ifLocal(txnId, homeKey, epoch, epoch, safeStore -> register(safeStore, txnId, homeKey)))
                 .begin(node.agent());
         }
 
@@ -278,7 +278,7 @@ class ExistingTransactionCheck extends AsyncResults.AbstractResult<BarrierTxn> i
         @Override
         public BarrierTxn apply(SafeCommandStore safeStore)
         {
-            // TODO (required): consider these semantics carefully
+            // TODO (required): it isn't probably safe to use a Read here; but, we can and should implement Barriers via writes
             // Barriers are trying to establish that committed transactions are applied before the barrier (or in this case just minEpoch)
             // so all existing transaction types should ensure that at this point. An earlier txnid may have an executeAt that is after
             // this barrier or the transaction we listen on and that is fine
@@ -289,8 +289,9 @@ public BarrierTxn apply(SafeCommandStore safeStore)
                 //noinspection SillyAssignment,ConstantConditions
                 safeStore = safeStore; // prevent use in lambda
                 safeStore.commandStore()
-                         .execute(found.txnId, safeStoreWithTxn -> register(safeStoreWithTxn, found.txnId, found.key))
-                         .begin(node.agent());
+                         .execute(found.txnId, safeStoreWithTxn -> {
+                             register(safeStoreWithTxn, found.txnId, found.key);
+                         }, node.agent());
             }
             return found;
         }

accord-core/src/main/java/accord/coordinate/CheckShards.java

@@ -67,7 +67,7 @@ protected CheckShards(Node node, TxnId txnId, U route, long srcEpoch, IncludeInf
 
     private static Topologies topologyFor(Node node, TxnId txnId, Unseekables<?> contact, long epoch)
     {
-        // TODO (expected): only fetch data from source epoch
+        // TODO (desired): only fetch data from source epoch
         return node.topology().preciseEpochs(contact, txnId.epoch(), epoch);
     }
 

accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java

@@ -32,6 +32,7 @@
 import accord.primitives.Txn;
 import accord.primitives.TxnId;
 import accord.topology.Topologies;
+import accord.utils.Invariants;
 import accord.utils.SortedListMap;
 import accord.utils.async.AsyncResult;
 import accord.utils.async.AsyncResults;
@@ -125,6 +126,7 @@ void onPreAccepted(Topologies topologies)
     {
         Deps deps = Deps.merge(oks, oks.domainSize(), SortedListMap::getValue, ok -> ok.deps);
         topologies = node.topology().reselect(topologies, QuorumEpochIntersections.preaccept.include, route, executeAtEpoch, executeAtEpoch, Owned);
-        new ExecuteEphemeralRead(node, topologies, route, txnId.withEpoch(executeAtEpoch), txn, executeAtEpoch, deps, this).start();
+        new ExecuteEphemeralRead(node, topologies, route, txnId.withEpoch(executeAtEpoch), txn, deps, this).start();
+        if (!Invariants.debug()) oks.clear();
     }
 }

accord-core/src/main/java/accord/coordinate/CoordinateGloballyDurable.java

@@ -31,12 +31,12 @@
 import accord.utils.async.AsyncResult;
 import accord.utils.async.AsyncResults.SettableResult;
 
-// TODO (expected): this does not need to query every shard; can disseminate globally any sub-range of the ring
-//  (indeed, we could slice both the query and dissemination only so that they always overlap)
+// TODO (expected): this does not need to query every shard OR more than one replica per shard;
+//  can disseminate globally any sub-range of the ring.
+//  Infact, we could simply autonomously disseminate our latest information to some subset of replicas
 public class CoordinateGloballyDurable extends SettableResult<Void> implements Callback<DurableBeforeReply>
 {
     final Node node;
-    // TODO (expected): this can be a ReadTracker, we only need one response from each shard
     final QuorumTracker tracker;
     private DurableBefore durableBefore = DurableBefore.EMPTY;
 

accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java

@@ -36,6 +36,7 @@
 import accord.primitives.Txn;
 import accord.primitives.TxnId;
 import accord.topology.Topologies;
+import accord.utils.Invariants;
 import accord.utils.SortedListMap;
 import accord.utils.WrappableException;
 
@@ -49,12 +50,10 @@
  * If we are preempted by a recovery coordinator, we abort and let them complete (and notify us about the execution result)
  *
  * TODO (desired, testing): dedicated burn test to validate outcomes
- * TODO (expected):
  */
 abstract class CoordinatePreAccept<T> extends AbstractCoordinatePreAccept<T, PreAcceptReply>
 {
     final PreAcceptTracker<?> tracker;
-    // TODO (expected): this can be cleared after preaccept
     private final SortedListMap<Id, PreAcceptOk> oks;
     final Txn txn;
     boolean fastPathEnabled = true;
@@ -148,10 +147,10 @@ void onNewEpochTopologyMismatch(TopologyMismatch mismatch)
     @Override
     void onPreAccepted(Topologies topologies)
     {
-        // TODO (expected): we do not have to take max here if we have enough fast path votes (this may unnecessarily force us onto the slow path)
         Timestamp executeAt = oks.foldlNonNullValues((ok, prev) -> mergeMaxAndFlags(ok.witnessedAt, prev), Timestamp.NONE);
         node.withEpoch(executeAt.epoch(), this, t -> WrappableException.wrap(t), () -> {
             onPreAccepted(topologies, executeAt, oks);
+            if (!Invariants.debug()) oks.clear();
         });
     }
 

accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java

@@ -56,6 +56,7 @@
 import static accord.primitives.Timestamp.Flag.HLC_BOUND;
 import static accord.primitives.Timestamp.Flag.REJECTED;
 import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
+import static accord.primitives.TxnId.Cardinality.cardinality;
 
 /**
  * Perform initial rounds of PreAccept and Accept until we have reached agreement about when we should execute.
@@ -124,14 +125,14 @@ public static <U extends Unseekable> AsyncResult<SyncPoint<U>> inclusiveAndAwait
     public static <U extends Unseekable> AsyncResult<SyncPoint<U>> coordinate(Node node, Kind kind, Unseekables<U> keysOrRanges, SyncPointAdapter<SyncPoint<U>> adapter)
     {
         Invariants.requireArgument(kind.isSyncPoint());
-        TxnId txnId = node.nextTxnId(kind, keysOrRanges.domain());
+        TxnId txnId = node.nextTxnId(kind, keysOrRanges.domain(), cardinality(keysOrRanges));
         return node.withEpoch(txnId.epoch(), () -> coordinate(node, txnId, keysOrRanges, adapter)).beginAsResult();
     }
 
     public static <U extends Unseekable> AsyncResult<SyncPoint<U>> coordinate(Node node, Kind kind, FullRoute<U> route, SyncPointAdapter<SyncPoint<U>> adapter)
     {
         Invariants.requireArgument(kind.isSyncPoint());
-        TxnId txnId = node.nextTxnId(kind, route.domain());
+        TxnId txnId = node.nextTxnId(kind, route.domain(), cardinality(route));
         return node.withEpoch(txnId.epoch(), () -> coordinate(node, txnId, route, adapter)).beginAsResult();
     }
 
@@ -150,7 +151,7 @@ private static <U extends Unseekable> AsyncResult<SyncPoint<U>> coordinate(Node
                                     : TopologyMismatch.checkForMismatchOrPendingRemoval(node.topology().globalForEpoch(txnId.epoch()), txnId, route.homeKey(), route);
         if (mismatch != null)
             return AsyncResults.failure(mismatch);
-        CoordinateSyncPoint<SyncPoint<U>> coordinate = new CoordinateSyncPoint<>(node, txnId, adapter.forDecision(node, route, txnId), node.agent().emptySystemTxn(txnId.kind(), txnId.domain()), route, adapter);
+        CoordinateSyncPoint<SyncPoint<U>> coordinate = new CoordinateSyncPoint<>(node, txnId, adapter.forDecision(node, route, txnId, txnId), node.agent().emptySystemTxn(txnId.kind(), txnId.domain()), route, adapter);
         coordinate.start();
         return coordinate;
     }

accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java

@@ -64,7 +64,7 @@
 import static accord.coordinate.Propose.NotAccept.proposeAndCommitInvalidate;
 import static accord.local.Commands.AcceptOutcome.Success;
 import static accord.primitives.Timestamp.Flag.REJECTED;
-import static accord.primitives.TxnId.FastPath.PRIVILEGED_COORDINATOR_WITH_DEPS;
+import static accord.primitives.TxnId.FastPath.PrivilegedCoordinatorWithDeps;
 import static java.util.concurrent.TimeUnit.MICROSECONDS;
 
 /**
@@ -148,7 +148,6 @@ protected CoordinationAdapter<Result> proposeAdapter()
         return Adapters.standard();
     }
 
-    // TODO (expected): override in C* rather than default to configurability here
     protected CoordinationAdapter<Result> executeAdapter()
     {
         return node.coordinationAdapter(txnId, Standard);
@@ -201,7 +200,7 @@ public void accept(PreAcceptReply result, Throwable failure)
                     // TODO (desired): we can probably still process and record fast path votes from peers, just with different quorum requirements
                     boolean hasCoordinatorVote = txnId.equals(ok.witnessedAt);
                     if (!hasCoordinatorVote) fastPathEnabled = false;
-                    Deps deps = hasCoordinatorVote && txnId.is(PRIVILEGED_COORDINATOR_WITH_DEPS) ? ok.deps : null;
+                    Deps deps = hasCoordinatorVote && txnId.is(PrivilegedCoordinatorWithDeps) ? ok.deps : null;
                     onSuccess(node.id(), ok);
                     for (Node.Id id : topologies.nodes())
                     {
@@ -225,7 +224,7 @@ public PreAcceptReply apply(SafeCommandStore safeStore)
             SafeCommand safeCommand = safeStore.get(txnId, participants);
 
             Deps deps;
-            if (txnId.is(PRIVILEGED_COORDINATOR_WITH_DEPS))
+            if (txnId.is(PrivilegedCoordinatorWithDeps))
             {
                 deps = PreAccept.calculateDeps(safeStore, txnId, participants, minEpoch, txnId, true);
                 if (deps == null)