Merge branch 'cassandra-6.0' into trunk

CHANGES.txt

@@ -3,6 +3,7 @@
  * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
  * Add a guardrail for misprepared statements (CASSANDRA-21139)
 Merged from 6.0:
+ * Safely regain ranges and delete retired command stores (CASSANDRA-21212)
  * Reduce memory allocations in miscellaneous places along read path (CASSANDRA-21360)
  * Avoid ByteBuffer wrapping in cql3.selection.Selector.InputRow to reduce memory allocation rate (CASSANDRA-21362)
  * Reduce cost to calculate BTreeRow.minDeletionTime (CASSANDRA-21414)

modules/accord

@@ -1 +1 @@
-Subproject commit 3d6392f4eca41d3036a704f1513020c24689f1bb
+Subproject commit 0a10cd056794c05588114f45ce86d49d6d6538db

src/java/org/apache/cassandra/db/compaction/CompactionIterator.java

@@ -1008,9 +1008,9 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
             if (read.kind() == Repeat && !hasWritten)
             {
                 Invariants.require(lastImage != null);
-                write = new TopologyImage(read.epoch(), Image, lastImage.getUpdate());
+                write = new TopologyImage(read.epoch(), Image, lastImage.update());
             }
-            else if (hasWritten && read.kind() == Repeat && lastImage.getUpdate().isEquivalent(read.getUpdate()))
+            else if (hasWritten && read.kind() == Repeat && lastImage.update().isEquivalent(read.update()))
             {
                 write = read.asRepeat();
             }

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

@@ -127,7 +127,9 @@ class AccordDurableOnFlush implements BiConsumer<Long, TableMetadata>
         for (Map.Entry<Integer, ReportDurable> e : notify.entrySet())
         {
             ReportDurable durable = e.getValue();
-            notifyInOrder(memtableId, metadata, commandStores.forId(e.getKey()), durable);
+            CommandStore commandStore = commandStores.forId(e.getKey());
+            if (commandStore != null)
+                notifyInOrder(memtableId, metadata, commandStore, durable);
         }
     }
 

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

@@ -40,6 +40,7 @@ import static org.apache.cassandra.db.TypeSizes.LONG_SIZE;
 import static org.apache.cassandra.service.accord.journal.MergeSerializers.BootstrapBeganAtSerializer;
 import static org.apache.cassandra.service.accord.journal.MergeSerializers.CommandChangeSerializer;
 import static org.apache.cassandra.service.accord.journal.MergeSerializers.DurableBeforeSerializer;
+import static org.apache.cassandra.service.accord.journal.MergeSerializers.PermanentlyUnsafeToReadSerializer;
 import static org.apache.cassandra.service.accord.journal.MergeSerializers.RangesForEpochSerializer;
 import static org.apache.cassandra.service.accord.journal.MergeSerializers.RedundantBeforeSerializer;
 import static org.apache.cassandra.service.accord.journal.MergeSerializers.SafeToReadSerializer;
@@ -273,7 +274,8 @@ public final class JournalKey
         SAFE_TO_READ                 (3, new SafeToReadSerializer(), false),
         BOOTSTRAP_BEGAN_AT           (4, new BootstrapBeganAtSerializer(), false),
         RANGES_FOR_EPOCH             (5, new RangesForEpochSerializer(), false),
-        TOPOLOGY_UPDATE              (6, new TopologySerializer(), true),
+        TOPOLOGY_UPDATE              (6, TopologySerializer.INSTANCE, true),
+        PERMANENTLY_UNSAFE_TO_READ   (7, new PermanentlyUnsafeToReadSerializer(), false)
         ;
 
         public final int id;

src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java

@@ -288,12 +288,13 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
                         logger.error("Encountered TopologyImage Repeat record for epoch {}, but no prior image record was found", ref.key().id.epoch());
                         return null;
                     }
-                    prev = reader.read().asImage(Invariants.nonNull(prev.getUpdate()));
+                    prev = reader.read().asImage(Invariants.nonNull(prev.update()));
                 }
                 else prev = reader.read();
 
-                return new accord.api.Journal.TopologyUpdate(prev.getUpdate().commandStores,
+                return new accord.api.Journal.TopologyUpdate(prev.update().commandStores,
-                                                             prev.getUpdate().global);
+                                                             prev.update().global,
+                                                             prev.update().previouslyOwned);
             }
 
             @Override
@@ -358,6 +359,13 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
         return accumulator.get();
     }
 
+    @Override
+    public Ranges loadPermanentlyUnsafeToRead(int commandStoreId)
+    {
+        KeepFirst<Ranges> accumulator = readLast(new JournalKey(TxnId.NONE, JournalKey.Type.PERMANENTLY_UNSAFE_TO_READ, commandStoreId));
+        return accumulator.get();
+    }
+
     @Override
     public PersistentField.Persister<DurableBefore, DurableBefore> durableBeforePersister()
     {
@@ -376,6 +384,8 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
             pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, commandStoreId), fieldUpdates.newSafeToRead);
         if (fieldUpdates.newRangesForEpoch != null)
             pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, commandStoreId), fieldUpdates.newRangesForEpoch);
+        if (fieldUpdates.newPermanentlyUnsafeToRead != null)
+            pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.PERMANENTLY_UNSAFE_TO_READ, commandStoreId), fieldUpdates.newPermanentlyUnsafeToRead);
 
         if (onFlush == null)
             return;

src/java/org/apache/cassandra/service/accord/journal/MergeSerializers.java

@@ -37,6 +37,7 @@ import org.apache.cassandra.service.accord.JournalKey;
 import org.apache.cassandra.service.accord.journal.Merger.KeepFirst;
 import org.apache.cassandra.service.accord.journal.Merger.SimpleMerger;
 import org.apache.cassandra.service.accord.serializers.CommandStoreSerializers;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
 import org.apache.cassandra.service.accord.serializers.Version;
 
 import static accord.local.CommandStores.RangesForEpoch;
@@ -250,7 +251,6 @@ public class MergeSerializers
                               SimpleMerger<?, ? super RangesForEpoch>,
                               SimpleMerger<RangesForEpoch, RangesForEpoch>>
     {
-        public static final RangesForEpochSerializer instance = new RangesForEpochSerializer();
         public KeepFirst<RangesForEpoch> mergerFor()
         {
             return new KeepFirst<>(null);
@@ -275,11 +275,40 @@ public class MergeSerializers
         }
     }
 
+    public static class PermanentlyUnsafeToReadSerializer
+    implements MergeSerializer<Ranges,
+                              SimpleMerger<?, ? super Ranges>,
+                              SimpleMerger<Ranges, Ranges>>
+    {
+        public KeepFirst<Ranges> mergerFor()
+        {
+            return new KeepFirst<>(Ranges.EMPTY);
+        }
+
+        @Override
+        public void serialize(JournalKey key, Ranges from, DataOutputPlus out, Version userVersion) throws IOException
+        {
+            KeySerializers.ranges.serialize(from, out);
+        }
+
+        @Override
+        public void reserialize(JournalKey key, SimpleMerger<Ranges, Ranges> from, DataOutputPlus out, Version userVersion) throws IOException
+        {
+            serialize(key, from.get(), out, userVersion);
+        }
+
+        @Override
+        public void deserialize(JournalKey key, SimpleMerger<?, ? super Ranges> into, DataInputPlus in, Version userVersion) throws IOException
+        {
+            into.update(KeySerializers.ranges.deserialize(in));
+        }
+    }
+
     public static class TopologySerializer implements MergeSerializer<TopologyRecord, TopologyMerger, TopologyMerger>
     {
         public static final TopologySerializer INSTANCE = new TopologySerializer();
 
-        public TopologySerializer() {}
+        private TopologySerializer() {}
 
         @Override
         public TopologyMerger mergerFor()
@@ -309,6 +338,7 @@ public class MergeSerializers
     public static class TopologyMerger implements Merger
     {
         TopologyRecord.TopologyImage read, write;
+        boolean hasRead;
 
         public TopologyMerger()
         {
@@ -317,6 +347,7 @@ public class MergeSerializers
         @Override
         public void reset(JournalKey key)
         {
+            hasRead = false;
             read = write = null;
         }
 
@@ -327,11 +358,15 @@ public class MergeSerializers
 
         public void read(TopologyRecord update)
         {
+            if (hasRead)
+                return;
+
             if (Objects.requireNonNull(update.kind()) == TopologyRecord.Kind.New)
-                read = new TopologyRecord.TopologyImage(update.epoch(), TopologyRecord.Kind.Image, update.getUpdate());
+                read = new TopologyRecord.TopologyImage(update.epoch(), TopologyRecord.Kind.Image, update.update());
             else
                 read = (TopologyRecord.TopologyImage) update;
             write = read;
+            hasRead = true;
         }
 
         public void write(TopologyRecord.TopologyImage image)

src/java/org/apache/cassandra/service/accord/journal/TopologyRecord.java

@@ -26,6 +26,7 @@ import org.agrona.collections.Int2ObjectHashMap;
 
 import accord.api.Journal;
 import accord.local.CommandStores;
+import accord.local.CommandStores.PreviouslyOwned;
 import accord.local.Node;
 import accord.primitives.Ranges;
 import accord.primitives.TxnId;
@@ -50,7 +51,7 @@ public interface TopologyRecord
     long epoch();
     TopologyRecord asRepeat();
 
-    Journal.TopologyUpdate getUpdate();
+    Journal.TopologyUpdate update();
     static TopologyRecord newTopology(Journal.TopologyUpdate update)
     {
         return new NewTopology(update);
@@ -121,7 +122,7 @@ public interface TopologyRecord
         }
 
         @Override
-        public Journal.TopologyUpdate getUpdate()
+        public Journal.TopologyUpdate update()
         {
             return update;
         }
@@ -182,7 +183,7 @@ public interface TopologyRecord
         }
 
         @Override
-        public Journal.TopologyUpdate getUpdate()
+        public Journal.TopologyUpdate update()
         {
             return update;
         }
@@ -225,12 +226,20 @@ public interface TopologyRecord
 
     class TopologyUpdateSerializer implements UnversionedSerializer<Journal.TopologyUpdate>
     {
+        private static final int TOP_BIT = 0x40000000;
         public static final TopologyUpdateSerializer instance = new TopologyUpdateSerializer();
 
         @Override
         public void serialize(Journal.TopologyUpdate from, DataOutputPlus out) throws IOException
         {
-            out.writeUnsignedVInt32(from.commandStores.size());
+            out.writeUnsignedVInt32(from.commandStores.size() | TOP_BIT);
+            out.writeUnsignedVInt32(0);
+            out.writeUnsignedVInt32(from.previouslyOwned.size());
+            for (int i = 0 ; i < from.previouslyOwned.size() ; ++i)
+            {
+                out.writeUnsignedVInt(from.previouslyOwned.epochs(i));
+                KeySerializers.ranges.serialize(from.previouslyOwned.ranges(i), out);
+            }
             for (Map.Entry<Integer, CommandStores.RangesForEpoch> e : from.commandStores.entrySet())
             {
                 out.writeUnsignedVInt32(e.getKey());
@@ -243,6 +252,23 @@ public interface TopologyRecord
         public Journal.TopologyUpdate deserialize(DataInputPlus in) throws IOException
         {
             int commandStoresSize = in.readUnsignedVInt32();
+            int flags = 0;
+            PreviouslyOwned previouslyOwned = PreviouslyOwned.EMPTY;
+            if ((commandStoresSize & TOP_BIT) != 0)
+            {
+                commandStoresSize ^= TOP_BIT;
+                // future proofing
+                flags = in.readUnsignedVInt32();
+                int previouslyOwnedSize = in.readUnsignedVInt32();
+                long[] epochs = new long[previouslyOwnedSize];
+                Ranges[] ranges = new Ranges[previouslyOwnedSize];
+                for (int i = 0 ; i < previouslyOwnedSize ; ++i)
+                {
+                    epochs[i] = in.readUnsignedVInt();
+                    ranges[i] = KeySerializers.ranges.deserialize(in);
+                }
+                previouslyOwned = new PreviouslyOwned(epochs.length == 0 ? 0 : epochs[0], epochs, ranges);
+            }
             Int2ObjectHashMap<CommandStores.RangesForEpoch> commandStores = new Int2ObjectHashMap<>();
             for (int j = 0; j < commandStoresSize; j++)
             {
@@ -251,13 +277,20 @@ public interface TopologyRecord
                 commandStores.put(commandStoreId, rfe);
             }
             Topology global = TopologySerializers.compactTopology.deserialize(in);
-            return new Journal.TopologyUpdate(commandStores, global);
+            return new Journal.TopologyUpdate(commandStores, global, previouslyOwned);
         }
 
         @Override
         public long serializedSize(Journal.TopologyUpdate from)
         {
-            long size = TypeSizes.sizeofUnsignedVInt(from.commandStores.size());
+            long size = TypeSizes.sizeofUnsignedVInt(from.commandStores.size() | TOP_BIT);
+            size += TypeSizes.sizeofUnsignedVInt(0);
+            size += TypeSizes.sizeofUnsignedVInt(from.previouslyOwned.size());
+            for (int i = 0 ; i < from.previouslyOwned.size() ; ++i)
+            {
+                size += TypeSizes.sizeofUnsignedVInt(from.previouslyOwned.epochs(i));
+                size += KeySerializers.ranges.serializedSize(from.previouslyOwned.ranges(i));
+            }
             for (Map.Entry<Integer, CommandStores.RangesForEpoch> e : from.commandStores.entrySet())
             {
                 size += TypeSizes.sizeofUnsignedVInt(e.getKey());

src/java/org/apache/cassandra/tcm/sequences/Move.java

@@ -27,12 +27,20 @@ import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import javax.annotation.Nullable;
+
 import com.google.common.annotations.VisibleForTesting;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import accord.api.TopologyListener;
+import accord.primitives.Ranges;
+import accord.topology.ActiveEpoch;
 import accord.topology.EpochReady;
+import accord.topology.Topology;
+import accord.topology.TopologyManager;
+import accord.topology.TopologyManager.RegainingEpochRange;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.SystemKeyspace;
@@ -54,6 +62,7 @@ import org.apache.cassandra.schema.ReplicationParams;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.accord.topology.AccordTopology;
 import org.apache.cassandra.streaming.StreamOperation;
 import org.apache.cassandra.streaming.StreamPlan;
 import org.apache.cassandra.streaming.StreamResultFuture;
@@ -75,10 +84,13 @@ import org.apache.cassandra.tcm.serialization.Version;
 import org.apache.cassandra.tcm.transformations.PrepareMove;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.concurrent.Condition;
 import org.apache.cassandra.utils.concurrent.Future;
 import org.apache.cassandra.utils.concurrent.FutureCombiner;
 import org.apache.cassandra.utils.vint.VIntCoding;
 
+import static accord.primitives.AbstractRanges.UnionMode.MERGE_ADJACENT;
+import static accord.primitives.Routables.Slice.Minimal;
 import static com.google.common.collect.ImmutableList.of;
 import static org.apache.cassandra.tcm.MultiStepOperation.Kind.MOVE;
 import static org.apache.cassandra.tcm.Transformation.Kind.FINISH_MOVE;
@@ -86,6 +98,7 @@ import static org.apache.cassandra.tcm.Transformation.Kind.MID_MOVE;
 import static org.apache.cassandra.tcm.Transformation.Kind.START_MOVE;
 import static org.apache.cassandra.tcm.sequences.SequenceState.continuable;
 import static org.apache.cassandra.tcm.sequences.SequenceState.error;
+import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition;
 
 public class Move extends MultiStepOperation<Epoch>
 {
@@ -196,6 +209,44 @@ public class Move extends MultiStepOperation<Epoch>
         return applyMultipleTransformations(metadata, next, of(startMove, midMove, finishMove));
     }
 
+    static class WaitForEpochAndRangeRetirement implements TopologyListener
+    {
+        final Condition condition = newOneTimeCondition();
+        final long waitingForEpoch;
+        final Ranges waitingForRanges;
+        Ranges retiredRanges;
+
+        public WaitForEpochAndRangeRetirement(long waitingForEpoch, Ranges waitingForRanges)
+        {
+            this.waitingForEpoch = waitingForEpoch;
+            this.waitingForRanges = waitingForRanges;
+            this.retiredRanges = Ranges.EMPTY;
+        }
+
+        synchronized void updateRetiredRanges(Ranges ranges)
+        {
+            ranges = ranges.slice(waitingForRanges, Minimal).without(retiredRanges);
+            if (!ranges.isEmpty())
+            {
+                retiredRanges = retiredRanges.union(MERGE_ADJACENT, ranges);
+                if (retiredRanges.containsAll(waitingForRanges))
+                    condition.signal();
+            }
+        }
+
+        @Override
+        public synchronized void onEpochRetired(Ranges ranges, long epoch, @Nullable Topology topology)
+        {
+            if (epoch >= waitingForEpoch)
+                updateRetiredRanges(ranges);
+        }
+
+        public void waitForRetirement()
+        {
+            condition.awaitThrowUncheckedOnInterrupt();
+        }
+    }
+
     @Override
     public Set<NodeId> affectedPeers(Directory directory)
     {
@@ -212,9 +263,38 @@ public class Move extends MultiStepOperation<Epoch>
         switch (next)
         {
             case START_MOVE:
+                WaitForEpochAndRangeRetirement wait = null;
+
                 try
                 {
                     ClusterMetadata metadata = ClusterMetadata.current();
+                    if (AccordService.isStarted() && metadata.schema.hasAccordKeyspaces())
+                    {
+                        TopologyManager topologyManager = AccordService.instance().topology();
+                        AccordService.toFuture(topologyManager.await(metadata.epoch.getEpoch(), null))
+                                     .awaitThrowUncheckedOnInterrupt().rethrowIfFailed();
+                        Topology current = topologyManager.active().globalForEpoch(metadata.epoch.getEpoch());
+                        RegainingEpochRange regaining = topologyManager.computeRegaining(current, AccordTopology.createAccordTopology(applyTo(metadata).success().metadata));
+
+                        if (regaining != null)
+                        {
+                            wait = new WaitForEpochAndRangeRetirement(regaining.epoch(), regaining.ranges());
+                            topologyManager.addListener(wait);
+                            ActiveEpoch e = topologyManager.active().ifExists(regaining.epoch());
+
+                            // We have already checked that our activeEpochs is caught up to metadata.epoch.getEpoch()
+                            // which is greater than regaining.epoch() so if e == null then the only case we can fall
+                            // in is that regaining.epoch() is already retired
+                            if (e != null)
+                            {
+                                wait.updateRetiredRanges(e.retired());
+                                logger.info("Waiting for previous ownership of ranges {} to retire before regaining", regaining.ranges());
+                                wait.waitForRetirement();
+                                logger.info("Previously owned ranges {} now retired", regaining.ranges());
+                            }
+                        }
+                    }
+
                     logger.info("Moving {} from {} to {}.",
                                 metadata.directory.endpoint(startMove.nodeId()),
                                 metadata.tokenMap.tokens(startMove.nodeId()),
@@ -226,6 +306,11 @@ public class Move extends MultiStepOperation<Epoch>
                     JVMStabilityInspector.inspectThrowable(t);
                     return continuable();
                 }
+                finally
+                {
+                    if (wait != null)
+                        AccordService.instance().topology().removeListener(wait);
+                }
                 break;
             case MID_MOVE:
                 try

test/distributed/org/apache/cassandra/distributed/test/accord/AccordDeleteCommandStoreTest.java

@@ -0,0 +1,81 @@
+/*
+ * 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.distributed.test.accord;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.Util;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.accord.AccordService;
+
+import static com.google.common.collect.Iterables.getOnlyElement;
+
+public class AccordDeleteCommandStoreTest extends AccordTestBase
+{
+    private static final Logger logger = LoggerFactory.getLogger(AccordDeleteCommandStoreTest.class);
+
+    @Override
+    protected Logger logger()
+    {
+        return logger;
+    }
+
+    @BeforeClass
+    public static void setupClass() throws IOException
+    {
+        AccordTestBase.setupCluster(builder -> builder
+                                               .withoutVNodes()
+                                               .withConfig(config ->
+                                                           config
+                                                           .set("accord.shard_durability_cycle", "20s")
+                                                           .set("accord.topology_sync_propagator_enabled_pre_start", true)
+                                                           .with(Feature.NETWORK, Feature.GOSSIP)), 4);
+    }
+
+    @Test
+    public void deleteCommandStoresTest() throws Throwable
+    {
+        List<String> ddls = Arrays.asList("DROP KEYSPACE IF EXISTS " + KEYSPACE + ';',
+                                          "CREATE KEYSPACE " + KEYSPACE + " WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 2}",
+                                          "CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'");
+        test(ddls, cluster -> {
+            String newToken = cluster.get(1).callOnInstance(() -> getOnlyElement(StorageService.instance.getTokens()));
+
+            int numberOfCommandStores = cluster.get(2).callOnInstance(() -> {
+                Util.spinUntilTrue(() -> AccordService.instance().node().commandStores().all().length > 0);
+                return AccordService.instance().node().commandStores().all().length;
+            });
+
+            cluster.get(2).runOnInstance(() -> {
+                StorageService.instance.move(Long.toString(Long.parseLong(newToken) + 1));
+                Util.spinUntilTrue(() -> AccordService.instance().node().commandStores().all().length < numberOfCommandStores,
+                                   60);
+            });
+        });
+    }
+}
+

test/distributed/org/apache/cassandra/distributed/test/accord/AccordRegainRangesTest.java

@@ -0,0 +1,138 @@
+/*
+ * 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.distributed.test.accord;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.local.CommandStore;
+import accord.local.CommandStores.PreviouslyOwned;
+import accord.local.PreLoadContext;
+import accord.primitives.AbstractRanges;
+import accord.primitives.Ranges;
+import accord.topology.ActiveEpochs;
+
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.SimpleQueryResult;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.TokenKey;
+
+import static com.google.common.collect.Iterables.getOnlyElement;
+import static org.apache.cassandra.service.accord.AccordService.getBlocking;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class AccordRegainRangesTest extends AccordTestBase
+{
+    private static final Logger logger = LoggerFactory.getLogger(AccordRegainRangesTest.class);
+
+    @Override
+    protected Logger logger()
+    {
+        return logger;
+    }
+
+    @BeforeClass
+    public static void setupClass() throws IOException
+    {
+        AccordTestBase.setupCluster(builder -> builder
+                                               .withoutVNodes()
+                                               .withConfig(config ->
+                                                           config
+                                                           .set("accord.shard_durability_target_splits", "1")
+                                                           .set("accord.shard_durability_cycle", "20s")
+                                                           .with(Feature.NETWORK, Feature.GOSSIP)), 6);
+    }
+
+    @Test
+    public void regainRangesTest() throws Throwable
+    {
+        List<String> ddls = Arrays.asList("DROP KEYSPACE IF EXISTS " + KEYSPACE + ';',
+                                          "CREATE KEYSPACE " + KEYSPACE + " WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 3}",
+                                          "CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'");
+        test(ddls, cluster -> {
+            cluster.coordinator(2).execute(wrapInTxn("INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (?, ?)"), ConsistencyLevel.SERIAL, 1, 2);
+
+            SimpleQueryResult result = cluster.coordinator(2).executeWithResult("SELECT token(k) FROM " + qualifiedAccordTableName + " WHERE k = 1 LIMIT 1", ConsistencyLevel.SERIAL);
+
+            String originalToken = cluster.get(2).callOnInstance(() -> getOnlyElement(StorageService.instance.getTokens()));
+
+            long token = (Long) result.toObjectArrays()[0][0];
+
+            assertTrue(token < Long.parseLong(originalToken));
+
+            // Loses range (-4069959284402364209,-3074457345618258603]
+            cluster.get(2).runOnInstance(() -> {
+                StorageService.instance.move(Long.toString(token - 1000));
+            });
+
+            // Regains range (-4069959284402364209,-3074457345618258603]
+            cluster.get(2).runOnInstance(() -> {
+                StorageService.instance.move(originalToken);
+            });
+
+            String tableName = accordTableName;
+
+            // Ensure no overlapping safeToRead ranges
+            cluster.get(2).runOnInstance(() -> {
+                TableId tid = Schema.instance.getTableMetadata(KEYSPACE, tableName).id();
+                TokenKey start = TokenKey.parse(tid, String.valueOf(token), Murmur3Partitioner.instance);
+                TokenKey end = TokenKey.parse(tid, originalToken, Murmur3Partitioner.instance);
+                Ranges regainedRange = Ranges.of(TokenRange.create(start, end));
+
+                ActiveEpochs activeEpochs = AccordService.instance().topology().active();
+                PreviouslyOwned previouslyOwned = AccordService.instance().node().commandStores().getPreviouslyOwnedFromSnapshot();
+                assertEquals(previouslyOwned.regains(regainedRange), regainedRange);
+                for (int i = 0; i < previouslyOwned.size(); i++)
+                {
+                    long epoch = previouslyOwned.epochs(i);
+                    if (!activeEpochs.hasAtLeastEpoch(epoch))
+                        continue;
+                    assertTrue(activeEpochs.getKnown(epoch).retired().containsAll(previouslyOwned.ranges(i)));
+                }
+
+                Ranges range = Ranges.EMPTY;
+                for (CommandStore commandStore : AccordService.instance().node().commandStores().all()) {
+                    Ranges safeToReadRanges = getBlocking(commandStore.submit((PreLoadContext.Empty) () -> "No overlapping safeToReadRanges", safeCommandStore -> {
+                        Ranges mergedRanges = Ranges.EMPTY;
+                        for (Ranges r : safeCommandStore.safeToReadAt().values())
+                            mergedRanges = mergedRanges.union(AbstractRanges.UnionMode.MERGE_ADJACENT, r);
+                        return mergedRanges;
+                    }));
+
+                    assertTrue(range.overlapping(safeToReadRanges).isEmpty());
+                    range = range.union(AbstractRanges.UnionMode.MERGE_ADJACENT, safeToReadRanges);
+                }
+            });
+        });
+    }
+}
+

test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java

@@ -92,10 +92,12 @@ public class AccordJournalCompactionTest
         NavigableMap<Timestamp, Ranges> safeToReadAtAccumulator = ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY);
         NavigableMap<TxnId, Ranges> bootstrapBeganAtAccumulator = ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY);
         RangesForEpoch rangesForEpochAccumulator = null;
+        Ranges permanentlyUnsafeToReadAccumulator = Ranges.EMPTY;
 
         Gen<DurableBefore> durableBeforeGen = AccordGenerators.durableBeforeGen(DatabaseDescriptor.getPartitioner());
         Gen<NavigableMap<Timestamp, Ranges>> safeToReadGen = AccordGenerators.safeToReadGen(DatabaseDescriptor.getPartitioner());
         Gen<RangesForEpoch> rangesForEpochGen = AccordGenerators.rangesForEpoch(DatabaseDescriptor.getPartitioner());
+        Gen<Ranges> permanentlyUnsafeToRead = AccordGenerators.ranges(DatabaseDescriptor.getPartitioner());
 
         AccordJournal journal = new AccordJournal(new TestParams()
         {
@@ -129,6 +131,7 @@ public class AccordJournalCompactionTest
 //                updates.addRedundantBefore = redundantBeforeGen.next(rs);
                 updates.newSafeToRead = safeToReadGen.next(rs);
                 updates.newRangesForEpoch = rangesForEpochGen.next(rs);
+                updates.newPermanentlyUnsafeToRead = permanentlyUnsafeToRead.next(rs);
 
                 journal.durableBeforePersister().persist(addDurableBefore, null);
                 journal.saveStoreState(1, updates, null);
@@ -141,6 +144,8 @@ public class AccordJournalCompactionTest
                     safeToReadAtAccumulator = updates.newSafeToRead;
                 if (updates.newRangesForEpoch != null)
                     rangesForEpochAccumulator = updates.newRangesForEpoch;
+                if (updates.newPermanentlyUnsafeToRead != null)
+                    permanentlyUnsafeToReadAccumulator = updates.newPermanentlyUnsafeToRead;
 
                 if (i % 100 == 0)
                     journal.closeCurrentSegmentForTestingIfNonEmpty();
@@ -153,6 +158,7 @@ public class AccordJournalCompactionTest
             Assert.assertEquals(bootstrapBeganAtAccumulator, journal.loadBootstrapBeganAt(1));
             Assert.assertEquals(safeToReadAtAccumulator, journal.loadSafeToRead(1));
             Assert.assertEquals(rangesForEpochAccumulator, journal.loadRangesForEpoch(1));
+            Assert.assertEquals(permanentlyUnsafeToReadAccumulator, journal.loadPermanentlyUnsafeToRead(1));
         }
         finally
         {

test/unit/org/apache/cassandra/service/accord/journal/TopologyRecordTest.java

@@ -40,6 +40,7 @@ import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.utils.AccordGenerators;
 
 import static accord.utils.Property.qt;
+import static org.apache.cassandra.service.accord.serializers.CommandStoreSerializersTest.previouslyOwnedGen;
 import static org.apache.cassandra.service.accord.serializers.CommandStoreSerializersTest.rangesForEpochGen;
 
 public class TopologyRecordTest
@@ -85,14 +86,13 @@ public class TopologyRecordTest
             Gen<Ranges> rangesGen = AccordGenerators.ranges(TBL1, partitioner);
             Gen<CommandStores.RangesForEpoch> rangesForEpochGen = rangesForEpochGen(rangesGen);
             Topology topology = AccordGenerators.topologyGen(rangesGen).next(rs);
+            CommandStores.PreviouslyOwned previouslyOwned = previouslyOwnedGen(rangesGen).next(rs);
 
             Int2ObjectHashMap<CommandStores.RangesForEpoch> commandStores = new Int2ObjectHashMap<>();
             for (Node.Id node : topology.nodes())
                 commandStores.put(node.id, rangesForEpochGen.next(rs));
 
-            Node.Id self = rs.pick(topology.nodes());
+            return new Journal.TopologyUpdate(commandStores, topology, previouslyOwned);
-
-            return new Journal.TopologyUpdate(commandStores, topology);
         };
     }
 
@@ -120,8 +120,8 @@ public class TopologyRecordTest
 
     private static void maybeUpdatePartitioner(TopologyRecord expected)
     {
-        Journal.TopologyUpdate update = expected.getUpdate();
+        Journal.TopologyUpdate update = expected.update();
         if (update != null)
-            maybeUpdatePartitioner(expected.getUpdate());
+            maybeUpdatePartitioner(expected.update());
     }
 }

test/unit/org/apache/cassandra/service/accord/serializers/CommandStoreSerializersTest.java

@@ -127,6 +127,27 @@ public class CommandStoreSerializersTest
         };
     }
 
+    public static Gen<CommandStores.PreviouslyOwned> previouslyOwnedGen(Gen<Ranges> rangesGen)
+    {
+        Gen.IntGen sizeGen = Gens.ints().between(0, 10);
+        Gen.LongGen epochGen = AccordGens.epochs();
+        return rs -> {
+            int size = sizeGen.nextInt(rs);
+            if (size == 0)
+                return CommandStores.PreviouslyOwned.EMPTY;
+            long maxEpoch = 0;
+            long[] epochs = new long[size];
+            Ranges[] ranges = new Ranges[size];
+            for (int i = 0; i < size; i++)
+            {
+                epochs[i] = epochGen.nextLong(rs);
+                ranges[i] = rangesGen.next(rs);
+                maxEpoch = Math.max(maxEpoch, epochs[i]);
+            }
+            return new CommandStores.PreviouslyOwned(maxEpoch, epochs, ranges);
+        };
+    }
+
     private void maybeUpdatePartitioner(CommandStores.RangesForEpoch expected)
     {
         if (expected.size() > 0)