FetchRequest should report as unavailable any slice that executes in a later epoch that is not owned by the replicas

Improve:
 - Remove GetMaxConflict; use local MaxConflict collection

Fix:
 - Invalidated should retain StoreParticipants so we can update CFK on journal replay
 - loading pruned uninitialised commands via CFK: make sure hasTouched contains key so that if invalidated we are notified
 - updateExecuteAtLeast should always be higher than TxnId
 - Async CFK callbacks treated pruned transactions incorrectly
 - node.withEpoch when ExecuteEphemeralRead in futureEpoch
 - Deps.without should be key/range aware
 - Durably mark bootstrapBeganAt and safeToReadAt in MaxConflicts
 - Don't attempt to calculate local deps when DepsErased in GetLatestDeps (command has durably applied to this shard)
 - filter StoreParticipants before invoking shouldCleanup
 - CFK should treat !stillTouches as outOfRange (rather than !touches)

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

Author: belliottsmith

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

@@ -39,6 +39,7 @@
 import accord.primitives.TxnId;
 import accord.topology.Topologies;
 import accord.utils.Invariants;
+import accord.utils.WrappableException;
 
 import static accord.coordinate.ReadCoordinator.Action.Aborted;
 import static accord.coordinate.ReadCoordinator.Action.Approve;
@@ -98,7 +99,9 @@ protected Action process(Id from, ReadReply reply)
             if (ok.futureEpoch > allTopologies.currentEpoch())
             {
                 // TODO (expected): only submit new requests for the keys that execute in a later epoch
-                new ExecuteEphemeralRead(node, node.topology().preciseEpochs(route, ok.futureEpoch, ok.futureEpoch), route, txnId.withEpoch(ok.futureEpoch), txn, ok.futureEpoch, deps, callback).start();
+                node.withEpoch(ok.futureEpoch, callback, t -> WrappableException.wrap(t), () -> {
+                    new ExecuteEphemeralRead(node, node.topology().preciseEpochs(route, ok.futureEpoch, ok.futureEpoch), route, txnId.withEpoch(ok.futureEpoch), txn, ok.futureEpoch, deps, callback).start();
+                });
                 return Aborted;
             }
 

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

@@ -342,17 +342,17 @@ private void recover()
         Invariants.checkState(committedExecuteAt == null || committedExecuteAt.equals(txnId));
         // should all be PreAccept
         Deps proposeDeps = LatestDeps.mergeProposal(recoverOkList, ok -> ok == null ? null : ok.deps);
-        Deps earlierAcceptedNoWitness = Deps.merge(recoverOkList, recoverOkList.size(), List::get, ok -> ok.earlierAcceptedNoWitness);
-        Deps earlierCommittedWitness = Deps.merge(recoverOkList, recoverOkList.size(), List::get, ok -> ok.earlierCommittedWitness);
-        earlierAcceptedNoWitness = earlierAcceptedNoWitness.without(earlierCommittedWitness::contains);
-        if (!earlierAcceptedNoWitness.isEmpty())
+        Deps earlierWait = Deps.merge(recoverOkList, recoverOkList.size(), List::get, ok -> ok.earlierWait);
+        Deps earlierNoWait = Deps.merge(recoverOkList, recoverOkList.size(), List::get, ok -> ok.earlierNoWait);
+        earlierWait = earlierWait.without(earlierNoWait);
+        if (!earlierWait.isEmpty())
         {
             // If there exist commands that were proposed an earlier execution time than us that have not witnessed us,
             // we have to be certain these commands have not successfully committed without witnessing us (thereby
             // ruling out a fast path decision for us and changing our recovery decision).
             // So, we wait for these commands to finish committing before retrying recovery.
             // See whitepaper for more details
-            awaitCommits(node, earlierAcceptedNoWitness).addCallback((success, failure) -> {
+            awaitCommits(node, earlierWait).addCallback((success, failure) -> {
                 if (failure != null) accept(null, failure);
                 else retry();
             });

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

@@ -34,11 +34,6 @@ public SimpleTracker(Topologies topologies, IntFunction<ST[]> arrayFactory, Func
         super(topologies, arrayFactory, trackerFactory);
     }
 
-    public SimpleTracker(Topologies topologies, IntFunction<ST[]> arrayFactory, ShardFactory<ST> trackerFactory)
-    {
-        super(topologies, arrayFactory, trackerFactory);
-    }
-
     public RequestStatus recordSuccess(Id from, boolean withFastPathTimestamp) { return recordSuccess(from); }
     public RequestStatus recordDelayed(Id from) { return NoChange; }
     public boolean hasFastPathAccepted() { return false; }

accord-core/src/main/java/accord/coordinate/tracking/SimpleTracker.java

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.Map;
 
+import accord.local.Command;
 import accord.local.SafeCommandStore;
 import accord.messages.ReadData;
 import accord.utils.async.AsyncChain;
@@ -50,8 +51,11 @@
 import accord.utils.async.AsyncChains;
 import accord.utils.async.AsyncResult;
 import accord.utils.async.AsyncResults;
+
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
+import static accord.messages.ReadEphemeralTxnData.retryInLaterEpoch;
 import static accord.primitives.SaveStatus.Applied;
 import static accord.primitives.SaveStatus.TruncatedApply;
 import static accord.messages.ReadData.CommitOrReadNack.Insufficient;
@@ -123,13 +127,8 @@ public CommandStore commandStore()
     }
 
     protected abstract PartialTxn rangeReadTxn(Ranges ranges);
-
     protected abstract void onReadOk(Node.Id from, CommandStore commandStore, Data data, Ranges ranges);
-
-    protected FetchRequest newFetchRequest(long sourceEpoch, TxnId syncId, Ranges ranges, PartialDeps partialDeps, PartialTxn partialTxn)
-    {
-        return new FetchRequest(sourceEpoch, syncId, ranges, partialDeps, rangeReadTxn(ranges));
-    }
+    protected abstract FetchRequest newFetchRequest(long sourceEpoch, TxnId syncId, Ranges ranges, PartialDeps partialDeps, PartialTxn partialTxn);
 
     @Override
     public void contact(Node.Id to, Ranges ranges)
@@ -187,7 +186,7 @@ public void onSuccess(Node.Id from, ReadReply reply)
                 }
 
                 // TODO (now): make sure it works if invoked in either order
-                inflight.remove(key).started(ok.maxApplied);
+                inflight.remove(key).started(ok.safeToReadAfter);
                 onReadOk(to, commandStore, ok.data, received);
                 // received must be invoked after submitting the persistence future, as it triggers onDone
                 // which creates a ReducingFuture over {@code persisting}
@@ -237,11 +236,16 @@ void abort(Ranges abort)
         // TODO (expected): implement abort
     }
 
-    public static class FetchRequest extends ReadData
+    public static abstract class FetchRequest extends ReadData
     {
+        // Note for future: we cannot safely execute on an Erased sync point without more work.
+        // Specifically, if the range has partially lost ownership on the recipient, the SyncPoint
+        // will not represent a safe point to snapshot from, and we won't have enough information to
+        // report the range as unavailable.
         private static final ExecuteOn EXECUTE_ON = new ExecuteOn(Applied, TruncatedApply);
         public final PartialTxn read;
         public final PartialDeps partialDeps;
+        private transient Timestamp safeToReadAfter;
 
         public FetchRequest(long sourceEpoch, TxnId syncId, Ranges ranges, PartialDeps partialDeps, PartialTxn partialTxn)
         {
@@ -268,22 +272,45 @@ protected AsyncChain<Data> beginRead(SafeCommandStore safeStore, Timestamp execu
             return read.read(safeStore, executeAt, unavailable);
         }
 
+        // must be invoked by implementations some time after the read has started OR must override safeToReadAt()
+        protected void readStarted(SafeCommandStore safeStore, Ranges unavailable)
+        {
+            safeToReadAfter = Timestamp.nonNullOrMax(Timestamp.NONE, Timestamp.nonNullOrMax(safeToReadAfter, safeStore.commandStore().unsafeGetMaxConflicts().foldl(Timestamp::nonNullOrMax)));
+        }
+
+        protected Timestamp safeToReadAfter()
+        {
+            return safeToReadAfter;
+        }
+
         @Override
         protected void readComplete(CommandStore commandStore, Data result, Ranges unavailable)
         {
-            Ranges reportUnavailable = unavailable.slice((Ranges)this.readScope, Minimal);
+            Ranges reportUnavailable = unavailable == null ? null : unavailable.slice((Ranges)this.readScope, Minimal);
             super.readComplete(commandStore, result, reportUnavailable);
         }
 
         @Override
         protected ReadOk constructReadOk(Ranges unavailable, Data data)
         {
-            return new FetchResponse(unavailable, data, maxApplied());
+            Timestamp safeToReadAfter = safeToReadAfter();
+            Invariants.checkState(data == null || safeToReadAfter != null);
+            return new FetchResponse(unavailable, data, safeToReadAfter);
         }
 
-        protected Timestamp maxApplied()
+        @Override
+        protected void read(SafeCommandStore safeStore, Command command)
         {
-            return null;
+            long retryInLaterEpoch = retryInLaterEpoch(executeAtEpoch, safeStore, command);
+            if (retryInLaterEpoch > 0)
+            {
+                Ranges unavailable = ((Ranges)readScope).slice(safeStore.ranges().allAt(executeAtEpoch), Minimal);
+                readComplete(safeStore.commandStore(), null, unavailable);
+            }
+            else
+            {
+                super.read(safeStore, command);
+            }
         }
 
         @Override
@@ -295,11 +322,13 @@ public MessageType type()
 
     public static class FetchResponse extends ReadOk
     {
-        public final @Nullable Timestamp maxApplied;
-        public FetchResponse(@Nullable Ranges unavailable, @Nullable Data data, @Nullable Timestamp maxApplied)
+        // only null if retryInFutureEpoch is set
+        public final @Nullable Timestamp safeToReadAfter;
+
+        public FetchResponse(@Nullable Ranges unavailable, @Nullable Data data, @Nonnull Timestamp safeToReadAfter)
         {
             super(unavailable, data);
-            this.maxApplied = maxApplied;
+            this.safeToReadAfter = safeToReadAfter;
         }
 
         @Override