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

modules/accord

@@ -1 +1 @@
-Subproject commit c5a984cfe41bb8d8f1d7a4cb446c194f829a5dd1
+Subproject commit bf85660dccaece9fcb3dd319b6f82265d780378f

src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java

@@ -431,7 +431,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
             SimpleDataSet ds = new SimpleDataSet(metadata());
             for (CommandStore commandStore : commandStores.all())
             {
-                DefaultProgressLog.ImmutableView view = (DefaultProgressLog.ImmutableView) commandStore.unsafeProgressLog();
+                DefaultProgressLog.ImmutableView view = ((DefaultProgressLog) commandStore.unsafeProgressLog()).immutableView();
                 TableId tableId = ((AccordCommandStore)commandStore).tableId();
                 TableMetadata tableMetadata = tableMetadata(tableId);
                 while (view.advance())

src/java/org/apache/cassandra/index/accord/RouteSSTableManager.java

@@ -28,8 +28,13 @@ import java.util.Map;
 import java.util.NavigableSet;
 import java.util.TreeSet;
 
+import org.apache.cassandra.io.FSReadError;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.utils.ByteArrayUtil;
+
+import static org.apache.cassandra.index.accord.IndexDescriptor.IndexComponent.CINTIA_SORTED_LIST;
 
 public class RouteSSTableManager implements SSTableManager
 {
@@ -85,7 +90,17 @@ public class RouteSSTableManager implements SSTableManager
         Group group = new Group(storeId, tableId);
         TreeSet<ByteBuffer> matches = new TreeSet<>();
         for (SSTableIndex index : sstables.values())
-            matches.addAll(index.search(group, start, startInclusive, end, endInclusive));
+        {
+            try
+            {
+                matches.addAll(index.search(group, start, startInclusive, end, endInclusive));
+            }
+            catch (Throwable t)
+            {
+                File file = index.id.fileFor(CINTIA_SORTED_LIST);
+                throw new FSReadError("Failed to search range index " + file + " for " + (startInclusive ? "[" : "(") + ByteArrayUtil.bytesToHex(start) + "..." + ByteArrayUtil.bytesToHex(end) + (endInclusive ? "]" : ")"), t, file);
+            }
+        }
         return matches;
     }
 

src/java/org/apache/cassandra/service/accord/AccordService.java

@@ -77,6 +77,7 @@ import accord.primitives.Status;
 import accord.primitives.Timestamp;
 import accord.primitives.Txn;
 import accord.primitives.TxnId;
+import accord.topology.Shard;
 import accord.topology.Topology;
 import accord.topology.TopologyManager;
 import accord.utils.DefaultRandom;
@@ -962,12 +963,13 @@ public class AccordService implements IAccordService, Shutdownable
     {
         // Need to make sure no existing txn are still being processed for this table... this is only used by DROP TABLE so NEW txn are expected to be blocked, so just need to "wait" for existing ones to complete
         Topology topology = node.topology().current();
-        List<TokenRange> rangeList = topology.reduce(new ArrayList<>(),
-                                                  s -> ((TokenRange) s.range).table().equals(id),
-                                                  (accum, s) -> {
-                                                      accum.add((TokenRange) s.range);
-                                                      return accum;
-                                                  });
+        List<TokenRange> rangeList = new ArrayList<>();
+        for (Shard shard : topology.shards())
+        {
+            TokenRange range = (TokenRange) shard.range;
+            if (id.equals(range.table()))
+                rangeList.add(range);
+        }
         if (rangeList.isEmpty()) return; // nothing to see here
 
         Ranges ranges = Ranges.of(rangeList.toArray(accord.primitives.Range[]::new));

src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java

@@ -63,10 +63,12 @@ public class AccordVerbHandler<T extends Request> implements IVerbHandler<T>
         Node.Id fromNodeId = endpointMapper.mappedId(message.from());
         long waitForEpoch = request.waitForEpoch();
         if (node.topology().hasAtLeastEpoch(waitForEpoch))
+        {
             request.process(node, fromNodeId, message.header);
+        }
         else
         {
-            node.withEpoch(waitForEpoch, (ignored, withEpochFailure) -> {
+            node.withEpochAtLeast(waitForEpoch, (ignored, withEpochFailure) -> {
                 if (withEpochFailure != null)
                     throw new RuntimeException("Timed out waiting for epoch when processing message from " + fromNodeId + " to " + node + " message " + message, withEpochFailure);
                 request.process(node, fromNodeId, message.header);

src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java

@@ -94,12 +94,12 @@ public class AccordInteropAdapter extends TxnAdapter
     }
 
     @Override
-    public void persist(Node node, Topologies any, Route<?> require, Route<?> sendTo, SelectNodeOwnership selectSendTo, FullRoute<?> route, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
+    public void persist(Node node, Topologies any, Route<?> require, Route<?> sendTo, SelectNodeOwnership selectSendTo, FullRoute<?> route, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, boolean informDurableOnDone, BiConsumer<? super Result, Throwable> callback)
     {
-        if (applyKind == Minimal && doInteropPersist(node, any, require, sendTo, selectSendTo, ballot, txnId, txn, executeAt, deps, writes, result, route, callback))
+        if (applyKind == Minimal && doInteropPersist(node, any, require, sendTo, selectSendTo, ballot, txnId, txn, executeAt, deps, writes, result, route, informDurableOnDone, callback))
             return;
 
-        super.persist(node, any, require, sendTo, selectSendTo, route, ballot, txnId, txn, executeAt, deps, writes, result, callback);
+        super.persist(node, any, require, sendTo, selectSendTo, route, ballot, txnId, txn, executeAt, deps, writes, result, informDurableOnDone, callback);
     }
 
 
@@ -116,7 +116,7 @@ public class AccordInteropAdapter extends TxnAdapter
         return true;
     }
 
-    private boolean doInteropPersist(Node node, Topologies any, Route<?> require, Route<?> sendTo, SelectNodeOwnership selectSendTo, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, BiConsumer<? super Result, Throwable> callback)
+    private boolean doInteropPersist(Node node, Topologies any, Route<?> require, Route<?> sendTo, SelectNodeOwnership selectSendTo, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, boolean informDurableOnDone, BiConsumer<? super Result, Throwable> callback)
     {
         Update update = txn.update();
         ConsistencyLevel consistencyLevel = update instanceof AccordUpdate ? ((AccordUpdate) update).cassandraCommitCL() : null;
@@ -124,7 +124,7 @@ public class AccordInteropAdapter extends TxnAdapter
             return false;
 
         Topologies all = execution(node, any, sendTo, selectSendTo, fullRoute, txnId, executeAt);
-        new AccordInteropPersist(node, all, txnId, require, ballot, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, callback)
+        new AccordInteropPersist(node, all, txnId, require, ballot, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, informDurableOnDone, callback)
             .start(Minimal, any, writes, result);
         return true;
     }

src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java

@@ -109,9 +109,9 @@ public class AccordInteropPersist extends Persist
     private final ConsistencyLevel consistencyLevel;
     private CallbackHolder callback;
 
-    public AccordInteropPersist(Node node, Topologies topologies, TxnId txnId, Route<?> sendTo, Ballot ballot, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ConsistencyLevel consistencyLevel, BiConsumer<? super Result, Throwable> clientCallback)
+    public AccordInteropPersist(Node node, Topologies topologies, TxnId txnId, Route<?> sendTo, Ballot ballot, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ConsistencyLevel consistencyLevel, boolean informDurableOnDone, BiConsumer<? super Result, Throwable> clientCallback)
     {
-        super(node, topologies, txnId, ballot, sendTo, txn, executeAt, deps, writes, result, fullRoute, AccordInteropApply.FACTORY);
+        super(node, topologies, txnId, ballot, sendTo, txn, executeAt, deps, writes, result, fullRoute, informDurableOnDone, AccordInteropApply.FACTORY);
         Invariants.requireArgument(consistencyLevel == ConsistencyLevel.QUORUM || consistencyLevel == ConsistencyLevel.ALL || consistencyLevel == ConsistencyLevel.SERIAL || consistencyLevel == ConsistencyLevel.ONE);
         this.consistencyLevel = consistencyLevel;
         registerClientCallback(result, clientCallback);

src/java/org/apache/cassandra/service/accord/serializers/GetEphmrlReadDepsSerializers.java

@@ -60,16 +60,21 @@ public class GetEphmrlReadDepsSerializers
         @Override
         public void serialize(GetEphemeralReadDepsOk reply, DataOutputPlus out) throws IOException
         {
-            DepsSerializers.deps.serialize(reply.deps, out);
             out.writeUnsignedVInt(reply.latestEpoch);
+            out.writeBoolean(reply.deps != null);
+            if (reply.deps == null)
+                return;
+            DepsSerializers.deps.serialize(reply.deps, out);
             out.writeUnsignedVInt32(reply.flags.bits());
         }
 
         @Override
         public GetEphemeralReadDepsOk deserialize(DataInputPlus in) throws IOException
         {
-            Deps deps = DepsSerializers.deps.deserialize(in);
             long latestEpoch = in.readUnsignedVInt();
+            if (!in.readBoolean())
+                return new GetEphemeralReadDepsOk(latestEpoch);
+            Deps deps = DepsSerializers.deps.deserialize(in);
             ExecuteFlags flags = ExecuteFlags.get(in.readUnsignedVInt32());
             return new GetEphemeralReadDepsOk(deps, latestEpoch, flags);
         }
@@ -77,9 +82,13 @@ public class GetEphmrlReadDepsSerializers
         @Override
         public long serializedSize(GetEphemeralReadDepsOk reply)
         {
-            return DepsSerializers.deps.serializedSize(reply.deps)
-                   + TypeSizes.sizeofUnsignedVInt(reply.latestEpoch)
-                   + TypeSizes.sizeofUnsignedVInt(reply.flags.bits());
+            long size = 1 + TypeSizes.sizeofUnsignedVInt(reply.latestEpoch);
+            if (reply.deps != null)
+            {
+                size += DepsSerializers.deps.serializedSize(reply.deps)
+                        + TypeSizes.sizeofUnsignedVInt(reply.flags.bits());
+            }
+            return size;
         }
     };
 }

test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java

@@ -29,6 +29,7 @@ import org.slf4j.LoggerFactory;
 import io.airlift.airline.Cli;
 import io.airlift.airline.Command;
 import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.simulator.SimulationRunner;
 import org.apache.cassandra.simulator.SimulatorUtils;
 import org.apache.cassandra.utils.StorageCompatibilityMode;
@@ -41,6 +42,7 @@ public class AccordSimulationRunner extends SimulationRunner
     public static void beforeAll()
     {
         CassandraRelevantProperties.JUNIT_STORAGE_COMPATIBILITY_MODE.setString(StorageCompatibilityMode.NONE.toString());
+        DatabaseDescriptor.clientInitialization();
     }
 
     @Command(name = "run")