CASSANDRA-21212 Deleted Retired Command Stores

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

@@ -28,6 +28,7 @@
 import accord.impl.CommandChange;
 import accord.local.Command;
 import accord.local.CommandStores;
+import accord.local.CommandStores.PreviouslyOwned;
 import accord.local.DurableBefore;
 import accord.local.Node;
 import accord.local.RedundantBefore;
@@ -85,11 +86,13 @@ class TopologyUpdate
     {
         public final Int2ObjectHashMap<CommandStores.RangesForEpoch> commandStores;
         public final Topology global;
+        public final PreviouslyOwned previouslyOwned;
 
-        public TopologyUpdate(@Nonnull Int2ObjectHashMap<CommandStores.RangesForEpoch> commandStores, @Nonnull Topology global)
+        public TopologyUpdate(@Nonnull Int2ObjectHashMap<CommandStores.RangesForEpoch> commandStores, @Nonnull Topology global, PreviouslyOwned previouslyOwned)
         {
             this.commandStores = commandStores;
             this.global = global;
+            this.previouslyOwned = previouslyOwned;
         }
 
         public boolean isEquivalent(TopologyUpdate other)
@@ -103,7 +106,7 @@ public boolean isEquivalent(TopologyUpdate other)
 
         public TopologyUpdate cloneWithEquivalentEpoch(long epoch)
         {
-            return new TopologyUpdate(commandStores, global.cloneEquivalentWithEpoch(epoch));
+            return new TopologyUpdate(commandStores, global.cloneEquivalentWithEpoch(epoch), previouslyOwned);
         }
 
         @Override
@@ -150,6 +153,7 @@ class FieldUpdates
         public RedundantBefore newRedundantBefore;
         public NavigableMap<TxnId, Ranges> newBootstrapBeganAt;
         public NavigableMap<Timestamp, Ranges> newSafeToRead;
+        public Ranges newPermanentlyUnsafeToRead;
         public CommandStores.RangesForEpoch newRangesForEpoch;
 
         public String toString()
@@ -161,6 +165,8 @@ public String toString()
                 builder.append("newBootstrapBeganAt=").append(newBootstrapBeganAt).append(", ");
             if (newSafeToRead != null)
                 builder.append("newSafeToRead=").append(newSafeToRead).append(", ");
+            if (newPermanentlyUnsafeToRead != null)
+                builder.append("newPermanentlyUnsafeToRead=").append(newPermanentlyUnsafeToRead).append(", ");
             if (newRangesForEpoch != null)
                 builder.append("newRangesForEpoch=").append(newRangesForEpoch).append(", ");
             builder.setLength(builder.length() - 2);

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

@@ -55,7 +55,6 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
         this.recoverIfAlreadyDurable = recoverIfAlreadyDurable;
         this.reportTo = reportTo;
     }
-
     public static Object maybeRecover(Node node, TxnId txnId, Infer.InvalidIf invalidIf, Route<?> someRoute, ProgressToken prevProgress, boolean recoverIfAlreadyDurable, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
     {
         MaybeRecover maybeRecover;

accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java

@@ -243,6 +243,12 @@ public final void setSafeToRead(NavigableMap<Timestamp, Ranges> newSafeToRead)
         ensureFieldUpdates().newSafeToRead = newSafeToRead;
     }
 
+    @Override
+    public final void setPermanentlyUnsafeToRead(Ranges newPermanentlyUnsafeToRead)
+    {
+        ensureFieldUpdates().newPermanentlyUnsafeToRead = newPermanentlyUnsafeToRead;
+    }
+
     @Override
     public void setRangesForEpoch(RangesForEpoch rangesForEpoch)
     {

accord-core/src/main/java/accord/local/CommandStore.java

@@ -187,6 +187,7 @@ CommandStore create(int id,
      * But they may still be ordered for other key ranges they participate in.
      */
     private NavigableMap<Timestamp, Ranges> safeToRead = emptySafeToRead();
+    private Ranges permanentlyUnsafeToRead = Ranges.EMPTY;
     private final Set<Bootstrap> bootstraps = Collections.synchronizedSet(new DeterministicIdentitySet<>());
     @Nullable private RejectBefore rejectBefore;
 
@@ -399,10 +400,17 @@ protected void unsafeAcceptRequests(Ranges accept)
     /**
      * This method may be invoked on a non-CommandStore thread
      */
-    final void unsafeSetSafeToRead(NavigableMap<Timestamp, Ranges> newSafeToRead)
+    final void unsafeSetSafeToRead(@Nullable NavigableMap<Timestamp, Ranges> newSafeToRead)
     {
+        if (newSafeToRead != null)
+            newSafeToRead = purgeHistory(newSafeToRead, permanentlyUnsafeToRead);
+        safeToRead = newSafeToRead;
         node.updateStamp();
-        this.safeToRead = newSafeToRead;
+    }
+
+    final void unsafeSetPermanentlyUnsafeToRead(Ranges newPermanentlyUnsafeToRead)
+    {
+        this.permanentlyUnsafeToRead = newPermanentlyUnsafeToRead;
     }
 
     protected final void unsafeClearSafeToRead()
@@ -1181,6 +1189,14 @@ final void markUnsafeToRead(Ranges ranges)
         }
     }
 
+    final AsyncChain<Void> markPermanentlyUnsafeToRead(Ranges ranges)
+    {
+        return chain((Empty) () -> "Mark Range As Permanently Unsafe To Read", safeStore -> {
+            safeStore.setSafeToRead(purgeHistory(safeToRead, ranges));
+            safeStore.setPermanentlyUnsafeToRead(permanentlyUnsafeToRead.union(MERGE_ADJACENT, ranges));
+        });
+    }
+
     public final DataStore unsafeGetDataStore()
     {
         return dataStore;