More follow-up to CASSANDRA-19967 and CASSANDRA-19869

Author: belliottsmith

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

@@ -59,10 +59,12 @@
 import accord.local.CommandStore;
 import accord.local.CommandStores.RangesForEpoch;
 import accord.local.Commands;
+import accord.local.DurableBefore;
 import accord.local.KeyHistory;
 import accord.local.Node;
 import accord.local.NodeTimeService;
 import accord.local.PreLoadContext;
+import accord.local.RedundantBefore;
 import accord.local.RedundantStatus;
 import accord.local.RejectBefore;
 import accord.local.SafeCommand;
@@ -397,7 +399,7 @@ public void run()
                 boolean done = command.hasBeen(Truncated);
                 if (!done)
                 {
-                    if (redundantBefore().status(txnId, command.route()) == RedundantStatus.PRE_BOOTSTRAP_OR_STALE)
+                    if (unsafeGetRedundantBefore().status(txnId, command.route()) == RedundantStatus.PRE_BOOTSTRAP_OR_STALE)
                         return;
 
                     Route<?> route = command.route().slice(allRanges);
@@ -759,7 +761,7 @@ protected void update(Command prev, Command updated)
                 return;
 
             Ranges slice = ranges(txnId, updated.executeAtOrTxnId());
-            slice = commandStore.redundantBefore().removeShardRedundant(txnId, updated.executeAtOrTxnId(), slice);
+            slice = commandStore.unsafeGetRedundantBefore().removeShardRedundant(txnId, updated.executeAtOrTxnId(), slice);
             commandStore.rangeCommands.computeIfAbsent(txnId, ignore -> new RangeCommand(commandStore.commands.get(txnId)))
                          .update(((AbstractRanges)updated.participants().touches()).toRanges().slice(slice, Minimal));
         }
@@ -801,6 +803,18 @@ public void setRangesForEpoch(RangesForEpoch rangesForEpoch)
             ranges = rangesForEpoch;
         }
 
+        @Override
+        public void upsertRedundantBefore(RedundantBefore addRedundantBefore)
+        {
+            unsafeUpsertRedundantBefore(addRedundantBefore);
+        }
+
+        @Override
+        public void upsertDurableBefore(DurableBefore addDurableBefore)
+        {
+            unsafeUpsertDurableBefore(addDurableBefore);
+        }
+
         @Override
         public NodeTimeService time()
         {

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

@@ -20,7 +20,7 @@
 
 import accord.api.RoutingKey;
 import accord.api.VisibleForImplementation;
-import accord.local.CommandStore;
+import accord.local.SafeCommandStore;
 import accord.primitives.RoutingKeys;
 import accord.primitives.Timestamp;
 import accord.primitives.TxnId;
@@ -38,15 +38,15 @@ public class TimestampsForKeys
 
     private TimestampsForKeys() {}
 
-    public static TimestampsForKey updateLastExecutionTimestamps(CommandStore commandStore, SafeTimestampsForKey tfk, TxnId txnId, Timestamp executeAt, boolean isForWriteTxn)
+    public static TimestampsForKey updateLastExecutionTimestamps(SafeCommandStore safeStore, SafeTimestampsForKey tfk, TxnId txnId, Timestamp executeAt, boolean isForWriteTxn)
     {
         TimestampsForKey current = tfk.current();
 
         Timestamp lastWrite = current.lastWriteTimestamp();
 
         if (executeAt.compareTo(lastWrite) < 0)
         {
-            if (commandStore.redundantBefore().preBootstrapOrStale(TxnId.min(txnId, current.lastWriteId()), RoutingKeys.of(tfk.key().toUnseekable())) == FULLY)
+            if (safeStore.redundantBefore().preBootstrapOrStale(TxnId.min(txnId, current.lastWriteId()), RoutingKeys.of(tfk.key().toUnseekable())) == FULLY)
                 return current;
             throw illegalState("%s is less than the most recent write timestamp %s", executeAt, lastWrite);
         }
@@ -59,7 +59,7 @@ public static TimestampsForKey updateLastExecutionTimestamps(CommandStore comman
 
         if (cmp < 0)
         {
-            if (!commandStore.safeToReadAt(executeAt).contains(tfk.key().toUnseekable()))
+            if (!safeStore.safeToReadAt(executeAt).contains(tfk.key().toUnseekable()))
                 return current;
             throw illegalState("%s is less than the most recent executed timestamp %s", executeAt, lastExecuted);
         }
@@ -83,6 +83,6 @@ public static TimestampsForKey updateLastExecutionTimestamps(CommandStore comman
     @VisibleForImplementation
     public static <D> TimestampsForKey updateLastExecutionTimestamps(AbstractSafeCommandStore<?,?,?> safeStore, RoutingKey key, TxnId txnId, Timestamp executeAt, boolean isForWriteTxn)
     {
-        return updateLastExecutionTimestamps(safeStore.commandStore(), safeStore.timestampsForKey(key), txnId, executeAt, isForWriteTxn);
+        return updateLastExecutionTimestamps(safeStore, safeStore.timestampsForKey(key), txnId, executeAt, isForWriteTxn);
     }
 }

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

@@ -52,16 +52,20 @@
 public enum Cleanup
 {
     NO(Uninitialised),
+    // we don't know if the command has been applied or invalidated as we have incomplete information
+    // so erase what information we don't need in future to decide this
+    // TODO (required): tighten up semantics here (and maybe infer more aggressively)
+    EXPUNGE_PARTIAL(TruncatedApplyWithOutcome),
     TRUNCATE_WITH_OUTCOME(TruncatedApplyWithOutcome),
     TRUNCATE(TruncatedApply),
     INVALIDATE(Invalidated),
     VESTIGIAL(ErasedOrVestigial),
     ERASE(Erased),
-    // erase all fields except any participants and committed executeAt
-    EXPUNGE_PARTIAL(Erased),
     // we can stop storing the record entirely
     EXPUNGE(Erased);
 
+    private static final Cleanup[] VALUES = values();
+
     public final SaveStatus appliesIfNot;
 
     Cleanup(SaveStatus appliesIfNot)
@@ -76,18 +80,19 @@ public final Cleanup filter(SaveStatus saveStatus)
 
     public static Cleanup shouldCleanup(SafeCommandStore safeStore, Command command)
     {
-        return shouldCleanup(safeStore.commandStore(), command, command.participants());
+        return shouldCleanup(safeStore, command, command.participants());
     }
 
     public static Cleanup shouldCleanup(SafeCommandStore safeStore, Command command, @Nonnull StoreParticipants participants)
     {
-        return shouldCleanup(safeStore.commandStore(), command, participants);
+        return shouldCleanup(command.txnId(), command.saveStatus(), command.durability(), participants,
+                             safeStore.redundantBefore(), safeStore.durableBefore());
     }
 
-    public static Cleanup shouldCleanup(CommandStore commandStore, Command command, @Nonnull StoreParticipants participants)
+    public static Cleanup shouldCleanup(Command command, RedundantBefore redundantBefore, DurableBefore durableBefore)
     {
-        return shouldCleanup(command.txnId(), command.saveStatus(), command.durability(), participants,
-                               commandStore.redundantBefore(), commandStore.durableBefore());
+        return shouldCleanup(command.txnId(), command.saveStatus(), command.durability(), command.participants(),
+                             redundantBefore, durableBefore);
     }
 
     public static Cleanup shouldCleanup(TxnId txnId, SaveStatus status, Durability durability, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
@@ -238,4 +243,9 @@ private static boolean expunge(TxnId txnId, SaveStatus saveStatus, DurableBefore
         //  then we can safely erase. Revisit as part of rationalising RedundantBefore registers.
         return redundantBefore.shardStatus(txnId) == SHARD_REDUNDANT;
     }
+
+    public static Cleanup forOrdinal(int ordinal)
+    {
+        return VALUES[ordinal];
+    }
 }

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

@@ -1368,7 +1368,7 @@ public static Update initialise(SafeCommandStore safeStore, TxnId txnId, Timesta
                         long maxEpoch = prevEpoch;
                         long epoch = rangesForEpoch.epochs[i];
                         Ranges ranges = rangesForEpoch.ranges[i];
-                        ranges = safeStore.commandStore().redundantBefore().removePreBootstrap(txnId, ranges);
+                        ranges = safeStore.redundantBefore().removePreBootstrap(txnId, ranges);
                         if (!ranges.isEmpty())
                         {
                             Unseekables<?> executionParticipants = participants.route.slice(ranges, Minimal);