Fixes

 - Don't assume stillExecutes applies to remote request - must mark unavailable any keys we don't have the txn definition for
 - Don't exit notifyManagedPreBootstrap notify loop early, as could have later transactions with lower txnId
 - CoordinateEphemeralRead must retry if insufficient responses from replicas that still own the range
 - Burn test terminates while non-recurring tasks pending on command store queues
 - Infinite recovery due to not sending InformDurable when partially truncated
 - Incorrect participants when invoking removeRedundantDependencies
 - Infinite bootstrap loop on retired topology
Improve
 - Do not perform linear filters of CommandStores or Topologies
 - Some default implementations of forEach/iterator
 - TopologyRetiredException messages

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

Author: belliottsmith

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

@@ -30,7 +30,6 @@
 import accord.topology.Topologies;
 import accord.utils.Invariants;
 import accord.utils.WrappableException;
-import accord.utils.async.AsyncResults.SettableResult;
 
 import static accord.api.ProtocolModifiers.QuorumEpochIntersections;
 import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
@@ -39,26 +38,28 @@
  * Abstract parent class for implementing preaccept-like operations where we may need to fetch additional replies
  * from future epochs.
  */
-abstract class AbstractCoordinatePreAccept<T, R> extends SettableResult<T> implements Callback<R>, BiConsumer<T, Throwable>
+abstract class AbstractCoordinatePreAccept<T, R> implements Callback<R>
 {
     final Node node;
     final TxnId txnId;
     final FullRoute<?> route;
 
     final Topologies topologies;
+    final BiConsumer<T, Throwable> callback;
     private boolean isDone;
 
-    AbstractCoordinatePreAccept(Node node, FullRoute<?> route, @Nonnull TxnId txnId)
+    AbstractCoordinatePreAccept(Node node, FullRoute<?> route, @Nonnull TxnId txnId, BiConsumer<T, Throwable> callback)
     {
-        this(node, route, txnId, node.topology().select(route, txnId, txnId, SHARE, QuorumEpochIntersections.preaccept.include));
+        this(node, route, txnId, node.topology().select(route, txnId, txnId, SHARE, QuorumEpochIntersections.preaccept.include), callback);
     }
 
-    AbstractCoordinatePreAccept(Node node, FullRoute<?> route, @Nonnull TxnId txnId, Topologies topologies)
+    AbstractCoordinatePreAccept(Node node, FullRoute<?> route, @Nonnull TxnId txnId, Topologies topologies, BiConsumer<T, Throwable> callback)
     {
         this.node = node;
         this.txnId = txnId;
         this.route = route;
         this.topologies = topologies;
+        this.callback = callback;
     }
 
     void start()
@@ -86,7 +87,8 @@ public final boolean onCallbackFailure(Id from, Throwable failure)
     {
         if (isDone) return false;
         isDone = true;
-        return tryFailure(failure);
+        callback.accept(null, failure);
+        return true;
     }
 
     @Override
@@ -103,19 +105,12 @@ public final void onSlowResponse(Id from)
             onSlowResponseInternal(from);
     }
 
-    @Override
-    public final boolean tryFailure(Throwable failure)
-    {
-        if (!super.tryFailure(failure))
-            return false;
-        onFailure(failure);
-        return true;
-    }
-
-    private void onFailure(Throwable failure)
+    void setFailure(Throwable failure)
     {
+        Invariants.require(!isDone);
         // we may already be complete, as we may receive a failure from a later phase; but it's fine to redundantly mark done
         isDone = true;
+        callback.accept(null, failure);
         if (failure instanceof CoordinationFailed)
         {
             ((CoordinationFailed) failure).set(txnId, route.homeKey());
@@ -133,7 +128,7 @@ final void onPreAcceptedOrNewEpoch()
         Invariants.require(!isDone);
         isDone = true;
         long latestEpoch = executeAtEpoch();
-        if (latestEpoch > topologies.currentEpoch()) node.withEpoch(latestEpoch, this, () -> onPreAcceptedInNewEpoch(topologies, latestEpoch));
+        if (latestEpoch > topologies.currentEpoch()) node.withEpochExact(latestEpoch, callback, t -> WrappableException.wrap(t), () -> onPreAcceptedInNewEpoch(topologies, latestEpoch));
         else onPreAccepted(topologies);
     }
 
@@ -143,11 +138,4 @@ final void onPreAcceptedInNewEpoch(Topologies topologies, long latestEpoch)
         if (mismatch == null) onPreAccepted(topologies);
         else onNewEpochTopologyMismatch(mismatch);
     }
-
-    @Override
-    public final void accept(T success, Throwable failure)
-    {
-        if (success != null) trySuccess(success);
-        else tryFailure(WrappableException.wrap(failure));
-    }
 }

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

@@ -19,6 +19,7 @@
 package accord.coordinate;
 
 import java.util.Collection;
+import java.util.function.BiConsumer;
 
 import accord.api.Result;
 import accord.coordinate.tracking.QuorumTracker;
@@ -35,7 +36,7 @@
 import accord.utils.Invariants;
 import accord.utils.SortedListMap;
 import accord.utils.async.AsyncResult;
-import accord.utils.async.AsyncResults;
+import accord.utils.async.AsyncResults.SettableByCallback;
 
 import static accord.api.ProtocolModifiers.QuorumEpochIntersections;
 import static accord.api.ProtocolModifiers.QuorumEpochIntersections.Include.Owned;
@@ -59,26 +60,36 @@
 public class CoordinateEphemeralRead extends AbstractCoordinatePreAccept<Result, GetEphemeralReadDepsOk>
 {
     public static AsyncResult<Result> coordinate(Node node, FullRoute<?> route, TxnId txnId, Txn txn)
+    {
+        SettableByCallback<Result> result = new SettableByCallback<>();
+        coordinate(node, route, txnId, txn, result);
+        return result;
+    }
+
+    public static void coordinate(Node node, FullRoute<?> route, TxnId txnId, Txn txn, BiConsumer<Result, Throwable> callback)
     {
         TopologyMismatch mismatch = TopologyMismatch.checkForMismatchOrPendingRemoval(node.topology().globalForEpoch(txnId.epoch()), txnId, route.homeKey(), route);
         if (mismatch != null)
-            return AsyncResults.failure(mismatch);
+        {
+            callback.accept(null, mismatch);
+            return;
+        }
 
         Topologies topologies = node.topology().withUnsyncedEpochs(route, txnId, txnId);
-        CoordinateEphemeralRead coordinate = new CoordinateEphemeralRead(node, topologies, route, txnId, txn);
+        CoordinateEphemeralRead coordinate = new CoordinateEphemeralRead(node, topologies, route, txnId, txn, callback);
         coordinate.start();
-        return coordinate;
     }
 
     private final Txn txn;
 
     private final QuorumTracker tracker;
     private final SortedListMap<Node.Id, GetEphemeralReadDepsOk> oks;
     private long executeAtEpoch;
+    private long retryInEpoch;
 
-    CoordinateEphemeralRead(Node node, Topologies topologies, FullRoute<?> route, TxnId txnId, Txn txn)
+    CoordinateEphemeralRead(Node node, Topologies topologies, FullRoute<?> route, TxnId txnId, Txn txn, BiConsumer<Result, Throwable> callback)
     {
-        super(node, route, txnId);
+        super(node, route, txnId, callback);
         this.txn = txn;
         this.tracker = new QuorumTracker(topologies);
         this.executeAtEpoch = txnId.epoch();
@@ -101,33 +112,50 @@ long executeAtEpoch()
     @Override
     public void onSuccessInternal(Node.Id from, GetEphemeralReadDepsOk ok)
     {
-        oks.put(from, ok);
-        if (ok.latestEpoch > executeAtEpoch)
-            executeAtEpoch = ok.latestEpoch;
-
-        if (tracker.recordSuccess(from) == Success)
-            onPreAcceptedOrNewEpoch();
+        if (ok.deps != null)
+        {
+            oks.put(from, ok);
+            if (ok.latestEpoch > executeAtEpoch)
+                executeAtEpoch = ok.latestEpoch;
+
+            if (tracker.recordSuccess(from) == Success)
+                onPreAcceptedOrNewEpoch();
+        }
+        else
+        {
+            retryInEpoch = ok.latestEpoch;
+            if (tracker.recordFailure(from) == Failed)
+                retry();
+        }
     }
 
     @Override
     public void onFailureInternal(Node.Id from, Throwable failure)
     {
         if (tracker.recordFailure(from) == Failed)
-            setFailure(new Timeout(txnId, route.homeKey()));
+        {
+            if (retryInEpoch > 0) retry();
+            else setFailure(new Timeout(txnId, route.homeKey()));
+        }
+    }
+
+    private void retry()
+    {
+        coordinate(node, route, txnId.withEpoch(retryInEpoch), txn, callback);
     }
 
     @Override
     void onNewEpochTopologyMismatch(TopologyMismatch mismatch)
     {
-        accept(null, mismatch);
+        setFailure(mismatch);
     }
 
     @Override
     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, SHARE, Owned);
-        new ExecuteEphemeralRead(node, topologies, route, txnId.withEpoch(executeAtEpoch), txn, deps, this).start();
+        new ExecuteEphemeralRead(node, topologies, route, txnId.withEpoch(executeAtEpoch), txn, deps, callback).start();
         if (!Invariants.debug()) oks.clear();
     }
 }

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

@@ -19,6 +19,7 @@
 package accord.coordinate;
 
 import java.util.Collection;
+import java.util.function.BiConsumer;
 
 import accord.api.VisibleForImplementation;
 import accord.coordinate.tracking.QuorumTracker;
@@ -32,6 +33,7 @@
 import accord.topology.Topologies;
 import accord.utils.async.AsyncResult;
 import accord.utils.async.AsyncResults;
+import accord.utils.async.AsyncResults.SettableByCallback;
 
 import static accord.coordinate.tracking.RequestStatus.Failed;
 import static accord.coordinate.tracking.RequestStatus.Success;
@@ -46,14 +48,14 @@ public class CoordinateMaxConflict extends AbstractCoordinatePreAccept<Timestamp
     Timestamp maxConflict;
     long executionEpoch;
 
-    private CoordinateMaxConflict(Node node, FullRoute<?> route, long executionEpoch)
+    private CoordinateMaxConflict(Node node, FullRoute<?> route, long executionEpoch, BiConsumer<Timestamp, Throwable> callback)
     {
-        this(node, route, executionEpoch, node.topology().withUnsyncedEpochs(route, executionEpoch, executionEpoch));
+        this(node, route, executionEpoch, node.topology().withUnsyncedEpochs(route, executionEpoch, executionEpoch), callback);
     }
 
-    private CoordinateMaxConflict(Node node, FullRoute<?> route, long executionEpoch, Topologies topologies)
+    private CoordinateMaxConflict(Node node, FullRoute<?> route, long executionEpoch, Topologies topologies, BiConsumer<Timestamp, Throwable> callback)
     {
-        super(node, route, null, topologies);
+        super(node, route, null, topologies, callback);
         this.maxConflict = Timestamp.NONE;
         this.executionEpoch = executionEpoch;
         this.tracker = new QuorumTracker(topologies);
@@ -66,9 +68,11 @@ public static AsyncResult<Timestamp> maxConflict(Node node, Routables<?> keysOrR
         TopologyMismatch mismatch = TopologyMismatch.checkForMismatchOrPendingRemoval(node.topology().globalForEpoch(epoch), null, route.homeKey(), keysOrRanges);
         if (mismatch != null)
             return AsyncResults.failure(mismatch);
-        CoordinateMaxConflict coordinate = new CoordinateMaxConflict(node, route, epoch);
+
+        SettableByCallback<Timestamp> result = new SettableByCallback<>();
+        CoordinateMaxConflict coordinate = new CoordinateMaxConflict(node, route, epoch, result);
         coordinate.start();
-        return coordinate;
+        return result;
     }
 
     @Override
@@ -91,13 +95,13 @@ void onSuccessInternal(Node.Id from, GetMaxConflictOk reply)
     void onFailureInternal(Node.Id from, Throwable failure)
     {
         if (tracker.recordFailure(from) == Failed)
-            tryFailure(failure);
+            setFailure(failure);
     }
 
     @Override
     void onNewEpochTopologyMismatch(TopologyMismatch mismatch)
     {
-        tryFailure(mismatch);
+        setFailure(mismatch);
     }
 
     @Override
@@ -109,6 +113,6 @@ long executeAtEpoch()
     @Override
     void onPreAccepted(Topologies topologies)
     {
-        setSuccess(maxConflict);
+        callback.accept(maxConflict, null);
     }
 }

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

@@ -19,6 +19,7 @@
 package accord.coordinate;
 
 import java.util.Collection;
+import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 
 import accord.coordinate.tracking.FastPathTracker;
@@ -59,19 +60,19 @@ abstract class CoordinatePreAccept<T> extends AbstractCoordinatePreAccept<T, Pre
     final Txn txn;
     boolean fastPathEnabled = true;
 
-    CoordinatePreAccept(Node node, TxnId txnId, Txn txn, FullRoute<?> route)
+    CoordinatePreAccept(Node node, TxnId txnId, Txn txn, FullRoute<?> route, BiConsumer<T, Throwable> callback)
     {
-        this(node, txnId, txn, route, node.topology().select(route, txnId, txnId, SHARE, QuorumEpochIntersections.preaccept.include));
+        this(node, txnId, txn, route, node.topology().select(route, txnId, txnId, SHARE, QuorumEpochIntersections.preaccept.include), callback);
     }
 
-    CoordinatePreAccept(Node node, TxnId txnId, Txn txn, FullRoute<?> route, Topologies topologies)
+    CoordinatePreAccept(Node node, TxnId txnId, Txn txn, FullRoute<?> route, Topologies topologies, BiConsumer<T, Throwable> callback)
     {
-        this(node, txnId, txn, route, topologies, FastPathTracker::new);
+        this(node, txnId, txn, route, topologies, FastPathTracker::new, callback);
     }
 
-    CoordinatePreAccept(Node node, TxnId txnId, Txn txn, FullRoute<?> route, Topologies topologies, BiFunction<Topologies, TxnId, PreAcceptTracker<?>> trackerFactory)
+    CoordinatePreAccept(Node node, TxnId txnId, Txn txn, FullRoute<?> route, Topologies topologies, BiFunction<Topologies, TxnId, PreAcceptTracker<?>> trackerFactory, BiConsumer<T, Throwable> callback)
     {
-        super(node, route, txnId, topologies);
+        super(node, route, txnId, topologies, callback);
         this.tracker = trackerFactory.apply(topologies, txnId);
         this.oks = new SortedListMap<>(topologies.nodes(), PreAcceptOk[]::new);
         this.txn = txn;
@@ -141,15 +142,15 @@ void onNewEpochTopologyMismatch(TopologyMismatch mismatch)
         proposeInvalidate(node, node.uniqueTimestamp(Ballot::fromValues), txnId, route.homeKey(), (outcome, failure) -> {
             if (failure != null)
                 mismatch.addSuppressed(failure);
-            accept(null, mismatch);
+            setFailure(mismatch);
         });
     }
 
     @Override
     void onPreAccepted(Topologies topologies)
     {
         Timestamp executeAt = oks.foldlNonNullValues((ok, prev) -> mergeMaxAndFlags(ok.witnessedAt, prev), Timestamp.NONE);
-        node.withEpoch(executeAt.epoch(), this, t -> WrappableException.wrap(t), () -> {
+        node.withEpochExact(executeAt.epoch(), callback, t -> WrappableException.wrap(t), () -> {
             onPreAccepted(topologies, executeAt, oks);
             if (!Invariants.debug()) oks.clear();
         });