Introduce Range transactions (#23)

patch by Benedict; reviewed by Ariel Weisberg for CASSANDRA-18174

Author: belliottsmith

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

@@ -30,6 +30,9 @@ public interface Key extends Seekable, RoutableKey
     @Override
     default Key asKey() { return this; }
 
+    @Override
+    default Key slice(Range range) { return this; }
+
     @Override
     default Range asRange() { throw new UnsupportedOperationException(); }
 }

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

@@ -19,10 +19,7 @@
 package accord.api;
 
 import accord.local.SafeCommandStore;
-import accord.primitives.Ranges;
-import accord.primitives.Keys;
-import accord.primitives.Timestamp;
-import accord.primitives.Txn;
+import accord.primitives.*;
 import org.apache.cassandra.utils.concurrent.AsyncPromise;
 import org.apache.cassandra.utils.concurrent.Future;
 
@@ -35,8 +32,8 @@
  */
 public interface Read
 {
-    Keys keys();
-    Future<Data> read(Key key, Txn.Kind kind, SafeCommandStore commandStore, Timestamp executeAt, DataStore store);
+    Seekables<?, ?> keys();
+    Future<Data> read(Seekable key, Txn.Kind kind, SafeCommandStore commandStore, Timestamp executeAt, DataStore store);
 
     class ReadFuture extends AsyncPromise<Data> implements BiConsumer<Data, Throwable>
     {

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

@@ -1,9 +1,16 @@
 package accord.api;
 
+import accord.primitives.Range;
 import accord.primitives.RoutableKey;
 import accord.primitives.Unseekable;
+import accord.utils.ArrayBuffers;
+
+import java.util.Arrays;
+
+import static accord.utils.ArrayBuffers.cachedRoutingKeys;
 
 public interface RoutingKey extends Unseekable, RoutableKey
 {
     @Override default RoutingKey toUnseekable() { return this; }
+    Range asRange();
 }

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

@@ -20,6 +20,7 @@
 
 import accord.primitives.Ranges;
 import accord.primitives.Keys;
+import accord.primitives.Seekables;
 
 import javax.annotation.Nullable;
 
@@ -30,7 +31,7 @@
  */
 public interface Update
 {
-    Keys keys();
+    Seekables<?, ?> keys();
     // null is provided only if nothing was read
     Write apply(@Nullable Data data);
     Update slice(Ranges ranges);

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

@@ -19,6 +19,7 @@
 package accord.api;
 
 import accord.local.SafeCommandStore;
+import accord.primitives.Seekable;
 import accord.primitives.Timestamp;
 import org.apache.cassandra.utils.concurrent.Future;
 
@@ -29,5 +30,5 @@
  */
 public interface Write
 {
-    Future<Void> apply(Key key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store);
+    Future<Void> apply(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store);
 }

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

@@ -40,6 +40,7 @@
 import static accord.local.PreLoadContext.contextFor;
 import static accord.local.SaveStatus.NotWitnessed;
 import static accord.local.Status.*;
+import static accord.primitives.Routables.Slice.Minimal;
 
 /**
  * Check on the status of a transaction. Returns early once enough information has been achieved to meet the requested
@@ -136,21 +137,23 @@ class OnDone implements MapReduceConsume<SafeCommandStore, Void>
 
         public OnDone()
         {
-            Ranges localRanges = node.topology().localRangesForEpochs(txnId.epoch(), untilLocalEpoch);
-            PartialRoute<?> selfRoute = route().slice(localRanges);
+            Ranges sliceRanges = node.topology().localRangesForEpochs(txnId.epoch(), untilLocalEpoch);
+            Ranges covering = route().sliceCovering(sliceRanges, Minimal);
+            Unseekables<?, ?> intersectingKeys = route().slice(covering, Minimal);
+
             full = (CheckStatusOkFull) merged;
-            sufficientFor = full.sufficientFor(selfRoute);
+            sufficientFor = full.sufficientFor(intersectingKeys);
             maxRoute = Route.merge((Route)route(), full.route);
             progressKey = node.trySelectProgressKey(txnId, maxRoute);
 
             PartialTxn partialTxn = null;
             if (sufficientFor.definition.isKnown())
-                partialTxn = full.partialTxn.slice(localRanges, true).reconstitutePartial(selfRoute);
+                partialTxn = full.partialTxn.slice(sliceRanges, true).reconstitutePartial(covering);
             this.partialTxn = partialTxn;
 
             PartialDeps partialDeps = null;
-            if (sufficientFor.deps.isDecisionKnown())
-                partialDeps = full.committedDeps.slice(localRanges).reconstitutePartial(selfRoute);
+            if (sufficientFor.deps.hasDecidedDeps())
+                partialDeps = full.committedDeps.slice(sliceRanges).reconstitutePartial(covering);
             this.partialDeps = partialDeps;
         }
 
@@ -161,7 +164,7 @@ void start()
                 keys = partialTxn.keys();
 
             Iterable<TxnId> txnIds = Collections.singleton(txnId);
-            if (sufficientFor.deps.isDecisionKnown())
+            if (sufficientFor.deps.hasDecidedDeps())
                 txnIds = Iterables.concat(txnIds, partialDeps.txnIds());
 
             PreLoadContext loadContext = contextFor(txnIds, keys);
@@ -219,7 +222,7 @@ public Void apply(SafeCommandStore safeStore)
             if (!safeStore.ranges().at(txnId.epoch()).contains(homeKey))
                 return null;
 
-            Timestamp executeAt = merged.saveStatus.known.executeAt.isDecisionKnown() ? merged.executeAt : null;
+            Timestamp executeAt = merged.saveStatus.known.executeAt.hasDecidedExecuteAt() ? merged.executeAt : null;
             command.setDurability(safeStore, merged.durability, homeKey, executeAt);
             safeStore.progressLog().durable(command, null);
             return null;

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

@@ -42,7 +42,8 @@ private static Topologies topologyFor(Node node, TxnId txnId, Unseekables<?, ?>
     @Override
     protected void contact(Id id)
     {
-        node.send(id, new CheckStatus(txnId, contact.slice(topologies().computeRangesForNode(id)), txnId.epoch(), untilRemoteEpoch, includeInfo), this);
+        Unseekables<?, ?> unseekables = contact.slice(topologies().computeRangesForNode(id));
+        node.send(id, new CheckStatus(txnId, unseekables, txnId.epoch(), untilRemoteEpoch, includeInfo), this);
     }
 
     protected boolean isSufficient(Id from, CheckStatusOk ok) { return isSufficient(ok); }

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

@@ -89,7 +89,7 @@ private static Object fetchInternal(Ranges ranges, Known target, Node node, TxnI
                 // information to fulfil that phase locally we should downgrade the response we give to the callback
                 Known sufficientFor = ok.sufficientFor(fetch);
                 // if we discover the executeAt as part of this action, use that to decide if we requested enough info
-                Timestamp exec = executeAt != null ? executeAt : ok.saveStatus.known.executeAt.isDecisionKnown() ? ok.executeAt : null;
+                Timestamp exec = executeAt != null ? executeAt : ok.saveStatus.known.executeAt.hasDecidedExecuteAt() ? ok.executeAt : null;
                 if (sufficientFor.outcome == OutcomeKnown && (exec == null || untilLocalEpoch < exec.epoch()))
                     sufficientFor = sufficientFor.with(OutcomeUnknown);
                 callback.accept(sufficientFor, null);

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

@@ -19,12 +19,15 @@
 package accord.coordinate;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.function.BiConsumer;
 
 import accord.coordinate.tracking.InvalidationTracker;
 import accord.coordinate.tracking.InvalidationTracker.InvalidationShardTracker;
 import accord.coordinate.tracking.RequestStatus;
+import accord.local.Node.Id;
 import accord.local.Status;
 import accord.messages.Commit;
 import accord.primitives.*;
@@ -59,6 +62,7 @@ public class Invalidate implements Callback<InvalidateReply>
     private final List<InvalidateReply> replies = new ArrayList<>();
     private final InvalidationTracker tracker;
     private Throwable failure;
+    private final Map<Id, InvalidateReply> debug = Invariants.debug() ? new HashMap<>() : null;
 
     private Invalidate(Node node, Ballot ballot, TxnId txnId, Unseekables<?, ?> invalidateWith, boolean transitivelyInvokedByPriorInvalidation, BiConsumer<Outcome, Throwable> callback)
     {
@@ -91,17 +95,18 @@ private void start()
     }
 
     @Override
-    public synchronized void onSuccess(Node.Id from, InvalidateReply reply)
+    public synchronized void onSuccess(Id from, InvalidateReply reply)
     {
         if (isDone || isPrepareDone)
             return;
 
+        if (debug != null) debug.put(from, reply);
         replies.add(reply);
         handle(tracker.recordSuccess(from, reply.isPromised(), reply.acceptedFastPath));
     }
 
     @Override
-    public void onFailure(Node.Id from, Throwable failure)
+    public void onFailure(Id from, Throwable failure)
     {
         if (isDone || isPrepareDone)
             return;
@@ -227,8 +232,9 @@ else if (homeKey != null)
 
         // if we have witnessed the transaction, but are able to invalidate, do we want to proceed?
         // Probably simplest to do so, but perhaps better for user if we don't.
-        // TODO (now, rangetxns): This should be a Routable, or we should guarantee it is safe to operate on any key in the range
-        RoutingKey invalidateWithKey = invalidateWith.slice(Ranges.of(tracker.promisedShard().range)).get(0).someIntersectingRoutingKey();
+        Ranges ranges = Ranges.of(tracker.promisedShard().range);
+        // we look up by TxnId at the target node, so it's fine to pick a RoutingKey even if it's a range transaction
+        RoutingKey invalidateWithKey = invalidateWith.slice(ranges).get(0).someIntersectingRoutingKey(ranges);
         proposeInvalidate(node, ballot, txnId, invalidateWithKey, (success, fail) -> {
             /*
               We're now inside our *exactly once* callback we registered with proposeInvalidate, and we need to
@@ -271,7 +277,7 @@ private void commitInvalidate()
     }
 
     @Override
-    public void onCallbackFailure(Node.Id from, Throwable failure)
+    public void onCallbackFailure(Id from, Throwable failure)
     {
         if (isDone)
             return;

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

@@ -19,7 +19,9 @@
 package accord.coordinate;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.function.BiConsumer;
 
 import accord.api.Result;
@@ -36,9 +38,11 @@
 import accord.local.Node.Id;
 import accord.messages.Accept;
 import accord.messages.Accept.AcceptReply;
+import accord.utils.Invariants;
 
 import static accord.coordinate.tracking.AbstractTracker.ShardOutcomes.Fail;
 import static accord.coordinate.tracking.RequestStatus.Failed;
+import static accord.utils.Invariants.debug;
 
 class Propose implements Callback<AcceptReply>
 {
@@ -50,6 +54,7 @@ class Propose implements Callback<AcceptReply>
     final Deps deps;
 
     private final List<AcceptReply> acceptOks;
+    private final Map<Id, AcceptReply> debug = debug() ? new HashMap<>() : null;
     private final Timestamp executeAt;
     private final QuorumTracker acceptTracker;
     private final BiConsumer<Result, Throwable> callback;
@@ -80,7 +85,7 @@ public static void propose(Node node, Topologies topologies, Ballot ballot, TxnI
                                Timestamp executeAt, Deps deps, BiConsumer<Result, Throwable> callback)
     {
         Propose propose = new Propose(node, topologies, ballot, txnId, txn, route, deps, executeAt, callback);
-        node.send(propose.acceptTracker.nodes(), to -> new Accept(to, topologies, ballot, txnId, route, executeAt, txn.keys(), deps, txn.kind()), propose);
+        node.send(propose.acceptTracker.nodes(), to -> new Accept(to, topologies, ballot, txnId, route, executeAt, txn.keys(), deps), propose);
     }
 
     @Override
@@ -89,6 +94,8 @@ public void onSuccess(Id from, AcceptReply reply)
         if (isDone)
             return;
 
+        if (debug != null) debug.put(from, reply);
+
         switch (reply.outcome())
         {
             default: throw new IllegalStateException();