Expose epoch ready state as a vtable for operators to inspect things

patch by David Capwell; reviewed by Benedict Elliott Smith for CASSANDRA-20302

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
-	url = https://github.com/ifesdjeen/cassandra-accord.git
-	branch = CASSANDRA-20297
+	url = https://github.com/apache/cassandra-accord.git
+	branch = trunk

modules/accord

@@ -1 +1 @@
-Subproject commit a2ac02b4d20bb6ab9078e70a0a43c17c00f4b0fc
+Subproject commit 74a3b81ca9d9e1ce7ddfd117682fc7c310f0cd99

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

@@ -17,32 +17,187 @@
  */
 package org.apache.cassandra.db.virtual;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
+import javax.annotation.Nullable;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
+
+import accord.primitives.Range;
+import accord.primitives.Ranges;
+import accord.topology.TopologyManager.EpochsSnapshot;
+import accord.topology.TopologyManager.EpochsSnapshot.Epoch;
+import accord.topology.TopologyManager.EpochsSnapshot.EpochReady;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
+import org.apache.cassandra.db.marshal.LongType;
+import org.apache.cassandra.db.marshal.ReversedType;
+import org.apache.cassandra.dht.LocalPartitioner;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.accord.TokenRange;
+
+import static accord.topology.TopologyManager.EpochsSnapshot.ResultStatus.SUCCESS;
 
 public class AccordVirtualTables
 {
-    private AccordVirtualTables() {}
+    public static final String EPOCHS = "accord_epochs";
+    public static final String TABLE_EPOCHS = "accord_table_epochs";
+
+    private AccordVirtualTables()
+    {
+    }
 
     public static Collection<VirtualTable> getAll(String keyspace)
     {
         if (!DatabaseDescriptor.getAccordTransactionsEnabled())
             return Collections.emptyList();
 
-        return List.of(
+        return List.of(new EpochReadyTable(keyspace),
+                       new EpochSyncRanges(keyspace)
         );
     }
 
-    private static TableMetadata parse(String keyspace, String comment, String query)
+    private static TableMetadata.Builder parse(String keyspace, String query)
     {
         return CreateTableStatement.parse(query, keyspace)
-                                   .comment(comment)
-                                   .kind(TableMetadata.Kind.VIRTUAL)
-                                   .build();
+                                   .kind(TableMetadata.Kind.VIRTUAL);
+    }
+
+    public static class EpochReadyTable extends AbstractVirtualTable
+    {
+        public EpochReadyTable(String keyspace)
+        {
+            super(parse(keyspace, "CREATE TABLE " + EPOCHS + " (\n" +
+                                  "  epoch bigint PRIMARY KEY,\n" +
+                                  "  ready_metadata text,\n" +
+                                  "  ready_coordinate text,\n" +
+                                  "  ready_data text,\n" +
+                                  "  ready_reads text,\n" +
+                                  "  ready boolean,\n" +
+                                  ")")
+                  .partitioner(new LocalPartitioner(ReversedType.getInstance(LongType.instance)))
+                  .comment("Exposes the epoch ready state for recieved epochs in Accord")
+                  .build());
+        }
+
+        @Override
+        public DataSet data()
+        {
+            SimpleDataSet ds = new SimpleDataSet(metadata());
+            EpochsSnapshot snapshot = epochsSnapshot();
+            for (Epoch epoch : snapshot)
+            {
+                ds.row(epoch.epoch);
+                EpochReady ready = epoch.ready;
+                ds.column("ready_metadata", ready.metadata.value);
+                ds.column("ready_coordinate", ready.coordinate.value);
+                ds.column("ready_data", ready.data.value);
+                ds.column("ready_reads", ready.reads.value);
+                ds.column("ready", ready.reads == SUCCESS);
+            }
+            return ds;
+        }
+    }
+
+    public static class EpochSyncRanges extends AbstractVirtualTable
+    {
+        protected EpochSyncRanges(String keyspace)
+        {
+            super(parse(keyspace, "CREATE TABLE " + TABLE_EPOCHS + " (\n" +
+                                  "  epoch bigint,\n" +
+                                  "  keyspace_name text,\n" +
+                                  "  table_name text,\n" +
+                                  "  added frozen<list<text>>,\n" +
+                                  "  removed frozen<list<text>>,\n" +
+                                  "  synced frozen<list<text>>,\n" +
+                                  "  closed frozen<list<text>>,\n" +
+                                  "  retired frozen<list<text>>,\n" +
+                                  "  PRIMARY KEY (epoch, keyspace_name, table_name)\n" +
+                                  ")")
+                  .partitioner(new LocalPartitioner(ReversedType.getInstance(LongType.instance)))
+                  .comment("Shows details on a per-table basis about what ranges are synced per epoch")
+                  .build());
+        }
+
+        @Override
+        public DataSet data()
+        {
+            SimpleDataSet ds = new SimpleDataSet(metadata());
+            EpochsSnapshot snapshot = epochsSnapshot();
+            for (Epoch state : snapshot)
+            {
+                Map<TableId, List<TokenRange>> addedRanges = groupByTable(state.addedRanges);
+                Map<TableId, List<TokenRange>> removedRanges = groupByTable(state.removedRanges);
+                Map<TableId, List<TokenRange>> synced = groupByTable(state.synced);
+                Map<TableId, List<TokenRange>> closed = groupByTable(state.closed);
+                Map<TableId, List<TokenRange>> retired = groupByTable(state.retired);
+
+                Set<TableId> allTables = union(addedRanges.keySet(), removedRanges.keySet(), synced.keySet(), closed.keySet(), retired.keySet());
+                for (TableId table : allTables)
+                {
+                    TableMetadata metadata = Schema.instance.getTableMetadata(table);
+                    if (metadata == null) continue; // table dropped, ignore
+                    ds.row(state.epoch, metadata.keyspace, metadata.name);
+
+                    ds.column("added", format(addedRanges.get(table)));
+                    ds.column("removed", format(removedRanges.get(table)));
+                    ds.column("synced", format(synced.get(table)));
+                    ds.column("closed", format(closed.get(table)));
+                    ds.column("retired", format(retired.get(table)));
+                }
+            }
+            return ds;
+        }
+
+        private static <T> Set<T> union(Set<T>... sets)
+        {
+            Preconditions.checkArgument(sets.length > 0);
+            if (sets.length == 1) return sets[0];
+            Sets.SetView accum = Sets.union(sets[0], sets[1]);
+            for (int i = 2; i < sets.length; i++)
+                accum = Sets.union(accum, sets[i]);
+            return accum;
+        }
+
+        private static List<String> format(@Nullable List<TokenRange> list)
+        {
+            if (list == null || list.isEmpty()) return Collections.emptyList();
+            List<String> result = new ArrayList<>(list.size());
+            for (TokenRange tr : list)
+                result.add(toStringNoTable(tr));
+            return result;
+        }
+    }
+
+    private static EpochsSnapshot epochsSnapshot()
+    {
+        return AccordService.instance().topology().epochsSnapshot();
+    }
+
+    private static String toStringNoTable(TokenRange tr)
+    {
+        // TokenRange extends Range.EndInclusive
+        return "(" + tr.start().suffix() + ", " + tr.end().suffix() + "]";
+    }
+
+    private static Map<TableId, List<TokenRange>> groupByTable(Ranges ranges)
+    {
+        Map<TableId, List<TokenRange>> map = new HashMap<>();
+        for (Range range : ranges)
+        {
+            TokenRange tr = (TokenRange) range;
+            map.computeIfAbsent(tr.table(), i -> new ArrayList<>()).add(tr);
+        }
+        return map;
     }
 }

test/distributed/org/apache/cassandra/distributed/api/QueryResults.java

@@ -25,6 +25,11 @@ import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.AbstractIterator;
 
 public final class QueryResults
 {
@@ -69,6 +74,34 @@ public final class QueryResults
         return new FilterQueryResult(result, fn);
     }
 
+    public static Iterable<List<String>> stringify(SimpleQueryResult qr)
+    {
+        return stringify(qr, -1);
+    }
+
+    public static Iterable<List<String>> stringify(SimpleQueryResult qr, int maxColumnSize)
+    {
+        Preconditions.checkArgument(maxColumnSize == -1 || maxColumnSize > 0, "max column size must be positive or -1 (disabled); given %s", maxColumnSize);
+        qr.mark();
+        return () -> {
+            qr.reset();
+            return new AbstractIterator<>()
+            {
+                @Override
+                protected List<String> computeNext()
+                {
+                    if (!qr.hasNext())
+                        return endOfData();
+                    Row next = qr.next();
+                    Stream<String> stream = Stream.of(next.toObjectArray()).map(Objects::toString);
+                    if (maxColumnSize != -1)
+                        stream = stream.map(s -> s.length() > maxColumnSize ? s.substring(0, maxColumnSize) + "..." : s);
+                    return stream.collect(Collectors.toList());
+                }
+            };
+        };
+    }
+
     public static Builder builder()
     {
         return new Builder();

test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java

@@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.stream.Collectors;
@@ -56,25 +57,30 @@ import org.apache.cassandra.cql3.ast.Txn;
 import org.apache.cassandra.db.marshal.AsciiType;
 import org.apache.cassandra.db.marshal.BytesType;
 import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.virtual.AccordVirtualTables;
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.distributed.api.IInstanceConfig;
 import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.QueryResults;
 import org.apache.cassandra.distributed.api.SimpleQueryResult;
 import org.apache.cassandra.distributed.shared.ClusterUtils;
 import org.apache.cassandra.distributed.test.accord.AccordTestBase;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.accord.api.AccordAgent;
 import org.apache.cassandra.service.consensus.TransactionalMode;
+import org.apache.cassandra.tools.nodetool.formatter.TableBuilder;
 import org.apache.cassandra.utils.ASTGenerators;
 import org.apache.cassandra.utils.AbstractTypeGenerators;
 import org.apache.cassandra.utils.CassandraGenerators;
 import org.apache.cassandra.utils.FastByteOperations;
 import org.apache.cassandra.utils.Generators;
 import org.apache.cassandra.utils.Isolated;
+import org.apache.cassandra.utils.Retry;
 import org.apache.cassandra.utils.Shared;
 import org.quicktheories.generators.SourceDSL;
 
+import static org.apache.cassandra.schema.SchemaConstants.VIRTUAL_VIEWS;
 import static org.apache.cassandra.utils.AbstractTypeGenerators.overridePrimitiveTypeSupport;
 import static org.apache.cassandra.utils.AbstractTypeGenerators.stringComparator;
 import static org.apache.cassandra.utils.AccordGenerators.fromQT;
@@ -121,7 +127,6 @@ public class AccordTopologyMixupTest extends TopologyMixupTestBase<AccordTopolog
     private static Spec createSchemaSpec(RandomSource rs, Cluster cluster)
     {
         TransactionalMode mode = rs.pick(TRANSACTIONAL_MODES);
-        boolean enableMigration = allowsMigration(mode) && rs.nextBoolean();
         // This test puts a focus on topology / cluster operations, so schema "shouldn't matter"... limit the domain of the test to improve the ability to debug
         AbstractTypeGenerators.TypeGenBuilder supportedTypes = AbstractTypeGenerators.withoutUnsafeEquality(AbstractTypeGenerators.builder()
                                                                                                                                   .withTypeKinds(AbstractTypeGenerators.TypeKind.PRIMITIVE));
@@ -140,12 +145,7 @@ public class AccordTopologyMixupTest extends TopologyMixupTestBase<AccordTopolog
         String schemaCQL = metadata.toCqlString(false, false, false);
         logger.info("Creating test table:\n{}", schemaCQL);
         cluster.schemaChange(schemaCQL);
-        if (enableMigration)
-        {
-            cluster.schemaChange("ALTER TABLE " + metadata + " WITH " + mode.asCqlParam());
-            cluster.get(1).nodetoolResult("consensus_admin", "begin-migration", metadata.keyspace, metadata.name).asserts().success();
-        }
-        return new Spec(mode, enableMigration, metadata);
+        return new Spec(mode, metadata);
     }
 
     private static CommandGen<Spec> cqlOperations(Spec spec)
@@ -175,18 +175,6 @@ public class AccordTopologyMixupTest extends TopologyMixupTestBase<AccordTopolog
         return new Property.SimpleCommand<>(node + ":" + msg + "; epoch=" + state.currentEpoch.get(), s2 -> executeTxn(s2.cluster, node, stmt.toCQL(), stmt.bindsEncoded()));
     }
 
-    private static boolean allowsMigration(TransactionalMode mode)
-    {
-        switch (mode)
-        {
-            case mixed_reads:
-            case full:
-                return true;
-            default:
-                return false;
-        }
-    }
-
     private static SimpleQueryResult executeTxn(Cluster cluster, IInvokableInstance node, String stmt, ByteBuffer[] binds)
     {
         if (!AccordTestBase.isIdempotent(node, stmt))
@@ -209,13 +197,11 @@ public class AccordTopologyMixupTest extends TopologyMixupTestBase<AccordTopolog
     public static class Spec implements Schema
     {
         private final TransactionalMode mode;
-        private final boolean enableMigration;
         private final TableMetadata metadata;
 
-        public Spec(TransactionalMode mode, boolean enableMigration, TableMetadata metadata)
+        public Spec(TransactionalMode mode, TableMetadata metadata)
         {
             this.mode = mode;
-            this.enableMigration = enableMigration;
             this.metadata = metadata;
         }
 
@@ -240,6 +226,8 @@ public class AccordTopologyMixupTest extends TopologyMixupTestBase<AccordTopolog
 
     private static class AccordState extends State<Spec>
     {
+        private final Map<Integer, String> instanceEpochReadyState = new TreeMap<>();
+        private final Map<Integer, String> instanceEpochSyncState = new TreeMap<>();
         private final ListenerHolder listener;
 
         public AccordState(RandomSource rs)
@@ -247,6 +235,36 @@ public class AccordTopologyMixupTest extends TopologyMixupTestBase<AccordTopolog
             super(rs, AccordTopologyMixupTest::createSchemaSpec, AccordTopologyMixupTest::cqlOperations);
 
             this.listener = new ListenerHolder(this);
+            this.preActions.add(this::populateEpochState);
+        }
+
+        private void populateEpochState()
+        {
+            updateMap(instanceEpochReadyState, "SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.EPOCHS);
+            updateMap(instanceEpochSyncState, "SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS);
+        }
+
+        private void updateMap(Map<Integer, String> map, String cql)
+        {
+            for (var inst : cluster)
+            {
+                int num = inst.config().num();
+                if (inst.isShutdown())
+                {
+                    map.put(num, "unknown");
+                    continue;
+                }
+                try
+                {
+                    SimpleQueryResult qr = Retry.retryWithBackoffBlocking(5, () -> cluster.get(num).executeInternalWithResult(cql));
+                    map.put(num, TableBuilder.toStringPiped(qr.names(), QueryResults.stringify(qr)));
+                }
+                catch (Throwable t)
+                {
+                    // Throwable.toString shows the type + msg but not the stack trace
+                    map.put(num, "unknown due to failure: " + t);
+                }
+            }
         }
 
         @Override
@@ -264,6 +282,19 @@ public class AccordTopologyMixupTest extends TopologyMixupTestBase<AccordTopolog
             ClusterUtils.awaitAccordEpochReady(cluster, tcmEpoch);
         }
 
+        @Override
+        public String toString()
+        {
+            StringBuilder sb = new StringBuilder(super.toString());
+            sb.append("\nAccord Epoch State:");
+            for (var e : instanceEpochReadyState.entrySet())
+                sb.append("\nnode").append(e.getKey()).append(":\n").append(e.getValue());
+            sb.append("\nAccord Epoch Ranges:");
+            for (var e : instanceEpochSyncState.entrySet())
+                sb.append("\nnode").append(e.getKey()).append(":\n").append(e.getValue());
+            return sb.toString();
+        }
+
         @Override
         public void close() throws Exception
         {

test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java

@@ -51,7 +51,7 @@ import org.agrona.collections.IntArrayList;
 import org.agrona.collections.IntHashSet;
 import org.apache.cassandra.distributed.Constants;
 import org.apache.cassandra.distributed.api.ICoordinator;
-import org.apache.cassandra.distributed.api.Row;
+import org.apache.cassandra.distributed.api.QueryResults;
 import org.apache.cassandra.distributed.api.SimpleQueryResult;
 
 import org.junit.Test;
@@ -374,30 +374,30 @@ public abstract class TopologyMixupTestBase<S extends TopologyMixupTestBase.Sche
         Property.StatefulBuilder statefulBuilder = stateful().withSteps(20).withStepTimeout(Duration.ofMinutes(3)).withExamples(1);
         preCheck(statefulBuilder);
         statefulBuilder.check(commands(this::stateGen)
-                .preCommands(state -> state.preActions.forEach(Runnable::run))
-                .add(2, (rs, state) -> {
-                    EnumSet<TopologyChange> possibleTopologyChanges = possibleTopologyChanges(state);
-                    if (possibleTopologyChanges.isEmpty()) return ignoreCommand();
-                    return topologyCommand(state, possibleTopologyChanges).next(rs);
-                })
-                .add(1, (rs, state) -> repairCommand(rs.pickInt(state.topologyHistory.up())))
-                .add(7, (rs, state) -> state.statementGen.apply(rs, state))
-                .destroyState((state, cause) -> {
-                    try (state)
-                    {
-                        TopologyMixupTestBase.this.destroyState(state, cause);
-                    }
-                })
-                .commandsTransformer((state, gen) -> {
-                    for (BiFunction<State<S>, Gen<Command<State<S>, Void, ?>>, Gen<Command<State<S>, Void, ?>>> fn : state.commandsTransformers)
-                        gen = fn.apply(state, gen);
-                    return gen;
-                })
-                .onSuccess((state, sut, history) -> logger.info("Successful for the following:\nState {}\nHistory:\n{}", state, Property.formatList("\t\t", history)))
-                .build());
+                              .preCommands(state -> state.preActions.forEach(Runnable::run))
+                              .addIf(State::allowTopologyChanges, 2, (rs, state) -> {
+                                  EnumSet<TopologyChange> possibleTopologyChanges = possibleTopologyChanges(state);
+                                  if (possibleTopologyChanges.isEmpty()) return ignoreCommand();
+                                  return topologyCommand(state, possibleTopologyChanges).next(rs);
+                              })
+                              .add(1, (rs, state) -> repairCommand(rs.pickInt(state.topologyHistory.up())))
+                              .add(7, (rs, state) -> state.statementGen.apply(rs, state))
+                              .destroyState((state, cause) -> {
+                                  try (state)
+                                  {
+                                      TopologyMixupTestBase.this.destroyState(state, cause);
+                                  }
+                              })
+                              .commandsTransformer((state, gen) -> {
+                                  for (BiFunction<State<S>, Gen<Command<State<S>, Void, ?>>, Gen<Command<State<S>, Void, ?>>> fn : state.commandsTransformers)
+                                      gen = fn.apply(state, gen);
+                                  return gen;
+                              })
+                              .onSuccess((state, sut, history) -> logger.info("Successful for the following:\nState {}\nHistory:\n{}", state, Property.formatList("\t\t", history)))
+                              .build());
     }
 
-    private EnumSet<TopologyChange> possibleTopologyChanges(State<S> state)
+    private static EnumSet<TopologyChange> possibleTopologyChanges(State<?> state)
     {
         EnumSet<TopologyChange> possibleTopologyChanges = EnumSet.noneOf(TopologyChange.class);
         // up or down is logically more correct, but since this runs sequentially and after the topology changes are complete, we don't have downed nodes at this point
@@ -683,6 +683,11 @@ public abstract class TopologyMixupTestBase<S extends TopologyMixupTestBase.Sche
             onStartupComplete(waitForEpoch);
         }
 
+        protected boolean allowTopologyChanges()
+        {
+            return !possibleTopologyChanges(this).isEmpty();
+        }
+
         protected void onStartupComplete(long tcmEpoch)
         {
 
@@ -753,17 +758,8 @@ public abstract class TopologyMixupTestBase<S extends TopologyMixupTestBase.Sche
             try
             {
                 SimpleQueryResult qr = Retry.retryWithBackoffBlocking(5, () -> cluster.get(cmsNode).executeInternalWithResult("SELECT epoch, kind, transformation FROM system_views.cluster_metadata_log"));
-                TableBuilder builder = new TableBuilder(" | ");
-                builder.add(qr.names());
-                while (qr.hasNext())
-                {
-                    Row next = qr.next();
-                    builder.add(Stream.of(next.toObjectArray())
-                            .map(Objects::toString)
-                            .map(s -> s.length() > 100 ? s.substring(0, 100) + "..." : s)
-                            .collect(Collectors.toList()));
-                }
-                epochHistory = "Epochs:\n" + builder;
+                String table = TableBuilder.toStringPiped(qr.names(), QueryResults.stringify(qr, 100));
+                epochHistory = "Epochs:\n" + table;
             }
             catch (Throwable t)
             {

test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java

@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.db.virtual;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import accord.api.ConfigurationService;
+import accord.api.TopologySorter;
+import accord.local.Node;
+import accord.primitives.Ranges;
+import accord.topology.Shard;
+import accord.topology.Topologies;
+import accord.topology.Topology;
+import accord.topology.TopologyManager;
+import accord.utils.SortedArrays;
+import accord.utils.async.AsyncResults;
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.accord.IAccordService;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.mockito.Mockito;
+
+import static org.apache.cassandra.schema.SchemaConstants.VIRTUAL_VIEWS;
+
+public class AccordVirtualTablesTest extends CQLTester
+{
+    public static final Node.Id N1 = new Node.Id(1);
+    public static final SortedArrays.SortedArrayList<Node.Id> ALL = SortedArrays.SortedArrayList.ofSorted(N1);
+    public static final Set<Node.Id> FP = Collections.singleton(N1);
+    public static final String SUCCESS = "success";
+    public static final String PENDING = "pending";
+    public static final List<String> FULL_RANGE = List.of("(-Inf, +Inf]");
+
+    public static TableId T1;
+    public static TableMetadata T1_META;
+
+    @BeforeClass
+    public static void setup()
+    {
+        addVirtualKeyspace();
+    }
+
+    @Before
+    public void setupTables()
+    {
+        if (T1_META != null) return;
+
+        String tbl1 = createTable("CREATE TABLE %s(pk int primary key)");
+        T1_META = Schema.instance.getTableMetadata(keyspace(), tbl1);
+        T1 = T1_META.id;
+
+        ServerTestUtils.markCMS();
+    }
+
+    @Test
+    public void emptyEpochs()
+    {
+        TopologyManager tm = empty();
+        assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.EPOCHS));
+        assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS));
+    }
+
+    @Test
+    public void epochUpdates()
+    {
+        TopologyManager tm = empty();
+        long e1 = 1;
+        tm.onTopologyUpdate(topology(e1, T1), () -> ConfigurationService.EpochReady.done(e1));
+        assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.EPOCHS),
+                   row(e1, true, SUCCESS, SUCCESS, SUCCESS, SUCCESS));
+
+        long e2 = 2;
+        tm.onTopologyUpdate(topology(e2, T1), () -> pendingReady(e1));
+        assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.EPOCHS),
+                   row(e2, false, PENDING, PENDING, PENDING, PENDING),
+                   row(e1, true, SUCCESS, SUCCESS, SUCCESS, SUCCESS));
+    }
+
+    @Test
+    public void tableUpdates()
+    {
+        TopologyManager tm = empty();
+        long e1 = 1;
+        tm.onTopologyUpdate(topology(e1, T1), () -> ConfigurationService.EpochReady.done(e1));
+
+        // the range was added in the first epoch, so its fully synced
+        assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS),
+                   row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, List.of(), List.of(), List.of(), FULL_RANGE));
+
+        // range is no longer "added" so doesn't show up as synced!
+        long e2 = 2;
+        tm.onTopologyUpdate(topology(e2, T1), () -> ConfigurationService.EpochReady.done(e2));
+        assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS),
+                   row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, List.of(), List.of(), List.of(), FULL_RANGE));
+
+        // sync the range
+        tm.onEpochSyncComplete(N1, e2);
+        assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS),
+                   row(e2, T1_META.keyspace, T1_META.name, List.of(), List.of(), List.of(), List.of(), FULL_RANGE),
+                   row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, List.of(), List.of(), List.of(), FULL_RANGE));
+
+        // lets close e2
+        tm.onEpochClosed(Ranges.single(TokenRange.fullRange(T1)), e2);
+        assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS),
+                   row(e2, T1_META.keyspace, T1_META.name, List.of(), FULL_RANGE, List.of(), List.of(), FULL_RANGE),
+                   row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, FULL_RANGE, List.of(), List.of(), FULL_RANGE));
+
+        // enjoy retirement!
+        tm.onEpochRetired(Ranges.single(TokenRange.fullRange(T1)), e2);
+        assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS),
+                   row(e2, T1_META.keyspace, T1_META.name, List.of(), FULL_RANGE, List.of(), FULL_RANGE, FULL_RANGE),
+                   row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, FULL_RANGE, List.of(), FULL_RANGE, FULL_RANGE));
+    }
+
+    private static ConfigurationService.EpochReady pendingReady(long epoch)
+    {
+        return new ConfigurationService.EpochReady(epoch, AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable());
+    }
+
+    private static Topology topology(long epoch, TableId tableId)
+    {
+        TokenRange all = TokenRange.fullRange(tableId);
+        return new Topology(epoch, Shard.create(all, ALL, FP));
+    }
+
+    private static TopologyManager empty()
+    {
+        TopologySorter sorter = (TopologySorter.StaticSorter) (node1, node2, shards) -> 0;
+        TopologySorter.Supplier supplier = new TopologySorter.Supplier()
+        {
+            @Override
+            public TopologySorter get(Topology topologies)
+            {
+                return sorter;
+            }
+
+            @Override
+            public TopologySorter get(Topologies topologies)
+            {
+                return sorter;
+            }
+        };
+        TopologyManager tm = new TopologyManager(supplier, null, N1, null, null, null);
+
+        var mock = Mockito.mock(IAccordService.class);
+        Mockito.when(mock.topology()).thenReturn(tm);
+        AccordService.unsafeSetNewAccordService(mock);
+        return tm;
+    }
+}

test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java

@@ -720,7 +720,7 @@ public class EpochSyncTest
                     @Override
                     public void onEpochRetired(Ranges ranges, long epoch)
                     {
-                        topology.onEpochRedundant(ranges, epoch);
+                        topology.onEpochRetired(ranges, epoch);
                     }
                 });