Add:

 - InMemoryRangeIndex
Fix:
 - DefaultLocalListeners iterators do not handle multiple waiters on the same transaction correctly
 - Minor burn test edge case when cancelling bootstrap and resuming
 - NotifyWaitingOn stillExecutes->stillWaitsOn
 - Privileged fast path commit failure should not send stable message
 - Erroneously populating executeAt with txnId in CommandSummaries for Accepted commands
Improve:
 - Improve Catchup logging
 - KeyDeps.forEach
 - Timestamp.tryParse
 - Simulate CFR loading/listening

Author: belliottsmith

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

@@ -127,7 +127,7 @@ static Spec parse(String description)
                         case "accept": accept = include; Invariants.require(cfp == DoNotChase, "Invalid to specify ChaseFixedPoint.Chase for accept"); break;
                         case "commit": commit = include; Invariants.require(cfp == DoNotChase, "Invalid to specify ChaseFixedPoint.Chase for commit"); break;
                         case "stable": stable = include; Invariants.require(cfp == DoNotChase, "Invalid to specify ChaseFixedPoint.Chase for stable"); break;
-                        case "recover": recover = include; Invariants.require(cfp == DoNotChase, "Invalid to specify ChaseFixedPoint.Chase for stable"); break;
+                        case "recover": recover = include; Invariants.require(cfp == DoNotChase, "Invalid to specify ChaseFixedPoint.Chase for recover"); break;
                     }
                 }
 
@@ -194,6 +194,32 @@ public static boolean discardPreAcceptDeps(TxnId txnId)
         }
     }
 
+    public static class RangeSpecParam
+    {
+        // Implementations may set this prior to creating any Range object to specify whether the start or end bound of its ranges are inclusive.
+        public static boolean END_INCLUSIVE = true;
+    }
+    public static class RangeSpec
+    {
+        public static final boolean END_INCLUSIVE = RangeSpecParam.END_INCLUSIVE;
+        public static boolean isEndInclusive()
+        {
+            return END_INCLUSIVE;
+        }
+        public static boolean isEndExclusive()
+        {
+            return !END_INCLUSIVE;
+        }
+        public static boolean isStartInclusive()
+        {
+            return !END_INCLUSIVE;
+        }
+        public static boolean isStartExclusive()
+        {
+            return END_INCLUSIVE;
+        }
+    }
+
     public static class Toggles
     {
         private static FastPaths permittedFastPaths = new FastPaths(FastPath.values());
@@ -205,10 +231,25 @@ public static class Toggles
         public static MediumPath defaultMediumPath() { return defaultMediumPath; }
         public static void setDefaultMediumPath(MediumPath newDefaultMediumPath) { defaultMediumPath = newDefaultMediumPath; }
 
+        private static boolean recoveryAwaitsSupersedingSyncPoints = true;
+        public static boolean recoveryAwaitsSupersedingSyncPoints() { return recoveryAwaitsSupersedingSyncPoints; }
+        public static void setRecoveryAwaitsSupersedingSyncPoints(boolean newRecoveryAwaitsSupersedingSyncPoints) { recoveryAwaitsSupersedingSyncPoints = newRecoveryAwaitsSupersedingSyncPoints; }
+
+        // TODO (required): default this to false once released support via recoveryAwaitsSupersedingSyncPoints
+        private static boolean syncPointsTrackUnstableMediumPathDependencies = true;
+        public static boolean syncPointsTrackUnstableMediumPathDependencies() { return syncPointsTrackUnstableMediumPathDependencies; }
+        public static void setSyncPointsTrackUnstableMediumPathDependencies(boolean newSyncPointsTrackUnstableMediumPathDependencies) { syncPointsTrackUnstableMediumPathDependencies = newSyncPointsTrackUnstableMediumPathDependencies; }
+
+        private static boolean recoverPartialAcceptPhaseIfNoFastPath = false;
+        public static boolean recoverPartialAcceptPhaseIfNoFastPath() { return recoverPartialAcceptPhaseIfNoFastPath; }
+        public static void setRecoverPartialAcceptPhaseIfNoFastPath(boolean newSyncPointsRecoverPartialAcceptPhase) {recoverPartialAcceptPhaseIfNoFastPath = newSyncPointsRecoverPartialAcceptPhase; }
+
         private static boolean filterDuplicateDependenciesFromAcceptReply = true;
         public static boolean filterDuplicateDependenciesFromAcceptReply() { return filterDuplicateDependenciesFromAcceptReply; }
         public static void setFilterDuplicateDependenciesFromAcceptReply(boolean newFilterDuplicateDependenciesFromAcceptReply) { filterDuplicateDependenciesFromAcceptReply = newFilterDuplicateDependenciesFromAcceptReply; }
 
+
+
         public enum SendStableMessages { TO_ALL, FOR_READS, FOR_READS_OR_NONE_IF_FASTEXEC}
         private static SendStableMessages sendStableMessages = FOR_READS_OR_NONE_IF_FASTEXEC;
         public static void setSendStableMessages(SendStableMessages newSendStableMessages) { sendStableMessages = newSendStableMessages; }

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

@@ -54,7 +54,6 @@
 
 import static accord.api.ProtocolModifiers.Toggles.permitLocalExecution;
 import static accord.coordinate.ExecutePath.EPHEMERAL;
-import static accord.coordinate.ExecutePath.FAST;
 import static accord.coordinate.ReadCoordinator.Action.Approve;
 import static accord.coordinate.ReadCoordinator.Action.ApprovePartial;
 import static accord.primitives.Status.Phase.Execute;

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

@@ -399,20 +399,23 @@ protected void onDone(Success success, Throwable failure)
         }
         else
         {
-            stable.informStableOnceQuorum();
-            if (sendOnlyReadStableMessages())
+            if (!isPrivilegedVoteCommitting)
             {
-                // send additional stable messages to record the transaction outcome
-                Commit.Kind kind = commitKind();
-                if (!candidates.isEmpty())
+                stable.informStableOnceQuorum();
+                if (sendOnlyReadStableMessages())
                 {
-                    for (int i = 0, size = candidates.size() ; i < size ; ++i)
-                        sendStableOnly(candidates.get(i), kind);
-                }
-                if (unstableFastReads != null)
-                {
-                    for (Node.Id to : unstableFastReads)
-                        sendStableOnly(to, kind);
+                    // send additional stable messages to record the transaction outcome
+                    Commit.Kind kind = commitKind();
+                    if (!candidates.isEmpty())
+                    {
+                        for (int i = 0, size = candidates.size() ; i < size ; ++i)
+                            sendStableOnly(candidates.get(i), kind);
+                    }
+                    if (unstableFastReads != null)
+                    {
+                        for (Node.Id to : unstableFastReads)
+                            sendStableOnly(to, kind);
+                    }
                 }
             }
             invokeCallback(null, failure);
@@ -430,9 +433,8 @@ public void onSlowResponse(Id from)
     @Override
     public void onFailure(Id from, Throwable failure)
     {
-        super.onFailure(from, failure);
-        if (isPrivilegedVoteCommitting && from.id == node.id().id)
-            tryFinishOnFailure();
+        if (isPrivilegedVoteCommitting && from.id == node.id().id) finishWithFailure(failure);
+        else super.onFailure(from, failure);
     }
 
     protected CoordinationAdapter<Result> adapter()

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

@@ -52,8 +52,10 @@ void start()
     {
         super.start();
         contact(ignore -> new GetDurableBefore(), id -> !node.id().equals(id));
-        markSelfContacted();
-        onSuccess(node.id(), new DurableBeforeReply(node.durableBefore()));
+        executor.executeMaybeImmediately(() -> {
+            markSelfContacted();
+            onSuccess(node.id(), new DurableBeforeReply(node.durableBefore()));
+        });
     }
 
     public static AsyncChain<DurableBefore> catchup(Node node)

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

@@ -24,6 +24,7 @@
 import org.slf4j.LoggerFactory;
 
 import accord.api.RoutingKey;
+import accord.api.VisibleForImplementation;
 import accord.local.CommandSummaries;
 import accord.local.MapReduceConsumeCommandStores;
 import accord.local.Node;
@@ -48,8 +49,6 @@
 import static accord.local.CommandSummaries.SummaryStatus.APPLIED;
 import static accord.local.CommandSummaries.SummaryStatus.COMMITTED;
 import static accord.local.CommandSummaries.SummaryStatus.INVALIDATED;
-import static accord.local.CommandSummaries.ComputeIsDep.IGNORE;
-import static accord.local.CommandSummaries.TestStartedAt.STARTED_AFTER;
 import static accord.local.durability.DurabilityService.SyncLocal.NoLocal;
 import static accord.local.durability.DurabilityService.SyncLocal.Self;
 import static accord.local.durability.DurabilityService.SyncRemote.NoRemote;
@@ -63,6 +62,7 @@
  * Facility for finding existing key transactions that can serve as a barrier transaction,
  * ensuring all reads/writes after some point in the txn log have been executed.
  */
+@VisibleForImplementation
 public class KeyBarriers
 {
     @SuppressWarnings("unused")
@@ -84,7 +84,7 @@ public Found(TxnId txnId, RoutingKey key, SyncLocal knownLocal, SyncRemote known
         }
     }
 
-    public static AsyncResult<Found> find(Node node, Timestamp min, RoutingKey key, SyncLocal syncLocal, SyncRemote syncRemote)
+    public static AsyncResult<Found> find(Node node, TxnId min, RoutingKey key, SyncLocal syncLocal, SyncRemote syncRemote)
     {
         Find find = new Find(min, key, syncLocal, syncRemote);
         node.commandStores().mapReduceConsume(min.epoch(), Long.MAX_VALUE, find);
@@ -98,16 +98,16 @@ public static AsyncResult<Found> find(Node node, Timestamp min, RoutingKey key,
      * For Applied we can return success immediately with the executeAt epoch. For PreApplied we can add
      * a listener for when it transitions to Applied and then return success.
      */
-    static class Find extends MapReduceConsumeCommandStores<RoutingKeys, Found> implements CommandSummaries.AllCommandVisitor
+    static class Find extends MapReduceConsumeCommandStores<RoutingKeys, Found> implements CommandSummaries.SupersedingCommandVisitor
     {
         final AsyncResults.SettableByCallback<Found> result = new AsyncResults.SettableByCallback<>();
-        final Timestamp min;
+        final TxnId min;
         final RoutingKey find;
         final SyncLocal syncLocal;
         final SyncRemote syncRemote;
         Found found;
 
-        Find(Timestamp min, RoutingKey find, SyncLocal syncLocal, SyncRemote syncRemote)
+        Find(TxnId min, RoutingKey find, SyncLocal syncLocal, SyncRemote syncRemote)
         {
             super(RoutingKeys.of(find));
             this.min = min;
@@ -122,7 +122,7 @@ public Found applyInternal(SafeCommandStore safeStore)
             // 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
-            safeStore.visit(scope, TxnId.NONE, Ws, STARTED_AFTER, min, IGNORE, this);
+            safeStore.visit(scope, min, Ws, this);
             return found;
         }
 

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

@@ -31,6 +31,7 @@
 import accord.local.Node.Id;
 import accord.local.SequentialAsyncExecutor;
 import accord.messages.Accept;
+import accord.messages.Accept.AcceptFlags;
 import accord.messages.Accept.AcceptReply;
 import accord.messages.Callback;
 import accord.primitives.Ballot;
@@ -47,9 +48,9 @@
 import accord.utils.SortedListMap;
 import accord.utils.Rethrowable;
 
-import static accord.api.ProtocolModifiers.Toggles.filterDuplicateDependenciesFromAcceptReply;
 import static accord.coordinate.tracking.RequestStatus.Failed;
 import static accord.coordinate.tracking.RequestStatus.Success;
+import static accord.messages.Accept.AcceptFlags.filterDeps;
 import static accord.messages.Commit.Invalidate.commitInvalidate;
 import static accord.primitives.Routables.Slice.Minimal;
 import static accord.primitives.Status.AcceptedInvalidate;
@@ -66,6 +67,7 @@ abstract class Propose<R> extends AbstractCoordination<FullRoute<?>, R, AcceptRe
 
     final Timestamp executeAt;
     final QuorumTracker tracker;
+    final int acceptFlags;
 
     Propose(Node node, SequentialAsyncExecutor executor, Topologies topologies, Accept.Kind kind, Ballot ballot, TxnId txnId, Txn txn, Route<?> require, FullRoute<?> route, Timestamp executeAt, Deps deps, BiConsumer<? super R, Throwable> callback)
     {
@@ -77,6 +79,7 @@ abstract class Propose<R> extends AbstractCoordination<FullRoute<?>, R, AcceptRe
         this.deps = deps;
         this.executeAt = executeAt;
         this.tracker = new QuorumTracker(topologies);
+        this.acceptFlags = AcceptFlags.encode(txnId, require != scope);
         Invariants.require(txnId.isSyncPoint() || deps.maxTxnId(txnId).compareTo(executeAt) <= 0,
                            "Attempted to propose %s with an earlier executeAt than a conflicting transaction it witnessed: %s vs executeAt: %s", txnId, deps, executeAt);
         Invariants.require(topologies.currentEpoch() == executeAt.epoch());
@@ -86,7 +89,7 @@ abstract class Propose<R> extends AbstractCoordination<FullRoute<?>, R, AcceptRe
     void start()
     {
         super.start();
-        contact(to -> new Accept(to, tracker.topologies(), kind, ballot, txnId, scope, executeAt, deps, require != scope));
+        contact(to -> new Accept(to, tracker.topologies(), kind, ballot, txnId, scope, executeAt, deps, acceptFlags));
     }
 
     @Override
@@ -163,15 +166,22 @@ Deps mergeDeps(Deps newDeps)
     Deps mergeNewDeps()
     {
         SortedListMap<Node.Id, AcceptReply> oks = finishOks();
+        if (!AcceptFlags.calculateDeps(acceptFlags))
+            return Deps.NONE;
+
         Deps deps = Deps.merge(oks, oks.domainSize(), SortedListMap::getValue, ok -> ok.deps);
         if (Faults.discardPreAcceptDeps(txnId))
             return deps;
 
         if (txnId.is(TrackStable))
         {
             // we must not propose as stable any dep < txnId that we did not propose as part of this phase
-            if (filterDuplicateDependenciesFromAcceptReply())
+            if (!filterDeps(acceptFlags))
+            {
+                // if the replicas haven't filtered their responses, we need to remove
+                // anything we sent to ensure we don't mark them as unstable
                 deps = deps.without(this.deps);
+            }
 
             deps = deps.markUnstableBefore(txnId);
         }
@@ -181,7 +191,6 @@ Deps mergeNewDeps()
 
     abstract CoordinationAdapter<R> adapter();
 
-
     @Override
     public CoordinationKind kind()
     {