Follow-up to CASSANDRA-19967 and CASSANDRA-19869

belliottsmith · belliottsmith · commit 4cf0070d604a · 2024-10-02T10:44:53.000+01:00
diff --git a/accord-core/src/main/java/accord/coordinate/Barrier.java b/accord-core/src/main/java/accord/coordinate/Barrier.java
@@ -182,7 +182,7 @@ private void createSyncPoint()
     {
         AsyncSyncPoint async = syncPoint.apply(node, route);
         coordinateSyncPoint = async.async;
-        if (!barrierType.global)
+        if (barrierType.async)
         {
             Invariants.checkState(barrierType.async);
             TxnId txnId = async.txnId;
diff --git a/accord-core/src/main/java/accord/impl/DefaultRequestTimeouts.java b/accord-core/src/main/java/accord/impl/DefaultRequestTimeouts.java
@@ -24,6 +24,7 @@
 
 import accord.api.RequestTimeouts;
 import accord.local.Node;
+import accord.utils.ArrayBuffers.BufferList;
 import accord.utils.LogGroupTimers;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -49,7 +50,8 @@ public void cancel()
                 lock.lock();
                 try
                 {
-                    timeouts.remove(this);
+                    if (isInHeap())
+                        timeouts.remove(this);
                 }
                 finally
                 {
@@ -91,16 +93,41 @@ public RegisteredTimeout tryRegister(Timeout timeout, long now, long deadline)
         @Override
         public void run()
         {
-            lock.lock();
-            try
+            try (BufferList<Registered> collect = new BufferList<>())
             {
-                long now = node.elapsed(MILLISECONDS);
-                // TODO (expected): should we handle reentrancy? Or at least throw an exception?
-                timeouts.advance(now, this, (s, r) -> r.timeout.timeout());
-            }
-            finally
-            {
-                lock.unlock();
+                int i = 0;
+                try
+                {
+                    lock.lock();
+                    try
+                    {
+                        long now = node.elapsed(MILLISECONDS);
+                        // TODO (expected): should we handle reentrancy? Or at least throw an exception?
+                        timeouts.advance(now, collect, BufferList::add);
+                    }
+                    finally
+                    {
+                        lock.unlock();
+                    }
+
+                    while (i < collect.size())
+                        collect.get(i++).timeout.timeout();
+                }
+                catch (Throwable t)
+                {
+                    while (i < collect.size())
+                    {
+                        try
+                        {
+                            collect.get(i++).timeout.timeout();
+                        }
+                        catch (Throwable t2)
+                        {
+                            t.addSuppressed(t2);
+                        }
+                    }
+                    throw t;
+                }
             }
         }
 
diff --git a/accord-core/src/main/java/accord/impl/ErasedSafeCommand.java b/accord-core/src/main/java/accord/impl/ErasedSafeCommand.java
@@ -37,8 +37,13 @@ public class ErasedSafeCommand extends SafeCommand
     public ErasedSafeCommand(TxnId txnId, SaveStatus saveStatus)
     {
         super(txnId);
+        this.erased = erased(txnId, saveStatus);
+    }
+
+    public static Command erased(TxnId txnId, SaveStatus saveStatus)
+    {
         Invariants.checkArgument(saveStatus.compareTo(Erased) >= 0);
-        this.erased = new Command.Truncated(txnId, saveStatus, saveStatus == ErasedOrVestigial ? NotDurable : UniversalOrInvalidated, StoreParticipants.empty(txnId), null, null, null);
+        return new Command.Truncated(txnId, saveStatus, saveStatus == ErasedOrVestigial ? NotDurable : UniversalOrInvalidated, StoreParticipants.empty(txnId), null, null, null);
     }
 
     @Override
diff --git a/accord-core/src/main/java/accord/impl/progresslog/HomeState.java b/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
@@ -132,8 +132,15 @@ final void runHome(DefaultProgressLog instance, SafeCommandStore safeStore, Safe
     {
         Invariants.checkState(!isHomeDoneOrUninitialised());
         Command command = safeCommand.current();
+        if (command.hasBeen(Status.Truncated))
+        {
+            // TODO (required): validate this better
+            setHomeDone(instance);
+            return;
+        }
         Invariants.checkState(!command.hasBeen(Status.Truncated), "Command %s is truncated", command);
 
+        Invariants.checkState(command.durability() != null);
         // TODO (expected): when invalidated, safer to maintain HomeState until known to be globally invalidated
         // TODO (now): validate that we clear HomeState when we receive a Durable reply, to replace the token check logic
         Invariants.checkState(!command.durability().isDurableOrInvalidated(), "Command is durable or invalidated, but we have not cleared the ProgressLog");
@@ -149,6 +156,13 @@ static void recoverCallback(SafeCommandStore safeStore, SafeCommand safeCommand,
         if (state == null)
             return;
 
+        Command command = safeCommand.current();
+        if (command.is(Status.Truncated))
+        {
+            state.setHomeDone(instance);
+            return;
+        }
+
         CoordinatePhase status = state.phase();
         if (status.isAtMostReadyToExecute() && state.homeProgress() == Querying)
         {
@@ -159,7 +173,6 @@ static void recoverCallback(SafeCommandStore safeStore, SafeCommand safeCommand,
             }
             else
             {
-                Command command = safeCommand.current();
                 ProgressToken token = success.asProgressToken().merge(command);
                 if (prevProgressToken != null)
                     token = token.merge(prevProgressToken);
diff --git a/accord-core/src/main/java/accord/local/Command.java b/accord-core/src/main/java/accord/local/Command.java
@@ -197,18 +197,18 @@ private AbstractCommand(TxnId txnId, SaveStatus status, Durability durability, @
         {
             this.txnId = txnId;
             this.status = validateCommandClass(txnId, status, getClass());
-            this.durability = durability;
-            this.participants = participants;
-            this.promised = promised;
+            this.durability = Invariants.nonNull(durability);
+            this.participants = Invariants.nonNull(participants);
+            this.promised = Invariants.nonNull(promised);
         }
 
         private AbstractCommand(CommonAttributes common, SaveStatus status, Ballot promised)
         {
             this.txnId = common.txnId();
             this.status = validateCommandClass(txnId, status, getClass());
-            this.durability = common.durability();
-            this.participants = common.participants();
-            this.promised = promised;
+            this.durability = Invariants.nonNull(common.durability());
+            this.participants = Invariants.nonNull(common.participants());
+            this.promised = Invariants.nonNull(promised);
         }
 
         @Override
@@ -1020,6 +1020,8 @@ private Committed(CommonAttributes common, SaveStatus status, Timestamp executeA
         {
             super(common, status, promised, executeAt, accepted);
             this.waitingOn = waitingOn;
+            Invariants.nonNull(common.participants());
+            Invariants.nonNull(common.route());
             Invariants.checkState(common.route().kind().isFullRoute(), "Expected a full route but given %s", common.route().kind());
         }
 
diff --git a/accord-core/src/main/java/accord/local/CommandStores.java b/accord-core/src/main/java/accord/local/CommandStores.java
@@ -52,6 +52,7 @@
 import accord.primitives.RoutingKeys;
 import accord.primitives.Timestamp;
 import accord.primitives.TxnId;
+import accord.topology.Shard;
 import accord.topology.Topology;
 import accord.utils.Invariants;
 import accord.utils.MapReduce;
@@ -452,8 +453,10 @@ private synchronized TopologyUpdate updateTopology(Node node, Snapshot prev, Top
             // TODO (desired): only sync affected shards
             Ranges ranges = shard.ranges().currentRanges();
             // ranges can be empty when ranges are lost or consolidated across epochs.
-            if (epoch > 1 && startSync && !ranges.isEmpty())
+            if (epoch > 1 && startSync && requiresSync(ranges, prev.global, newTopology))
+            {
                 bootstrapUpdates.add(shard.store.sync(node, ranges, epoch));
+            }
             result.add(shard);
         }
 
@@ -489,6 +492,37 @@ private synchronized TopologyUpdate updateTopology(Node node, Snapshot prev, Top
         return new TopologyUpdate(new Snapshot(result.toArray(new ShardHolder[0]), newLocalTopology, newTopology), bootstrap);
     }
 
+    private static boolean requiresSync(Ranges ranges, Topology oldTopology, Topology newTopology)
+    {
+        List<Shard> oldShards = oldTopology.foldl(ranges, (oldShard, shards, i) -> {
+            shards.add(oldShard);
+            return shards;
+        }, new ArrayList<>());
+
+        List<Shard> newShards = newTopology.foldl(ranges, (newShard, shards, i) -> {
+            shards.add(newShard);
+            return shards;
+        }, new ArrayList<>());
+
+        if (oldShards.size() != newShards.size())
+            return true;
+
+        for (int i = 0 ; i < oldShards.size() ; ++i)
+        {
+            Shard oldShard = oldShards.get(i);
+            Shard newShard = newShards.get(i);
+            if (!oldShard.fastPathElectorate.containsAll(newShard.fastPathElectorate))
+                return true;
+
+            if (!newShard.fastPathElectorate.containsAll(oldShard.fastPathElectorate))
+                return true;
+
+            if (!newShard.nodes.equals(oldShard.nodes))
+                return true;
+        }
+        return false;
+    }
+
     public <R> R unsafeFoldLeft(R initial, BiFunction<R, CommandStore, R> f)
     {
         Snapshot snapshot = current;
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
@@ -605,14 +605,12 @@ public static boolean maybeExecute(SafeCommandStore safeStore, SafeCommand safeC
                                                                     .intersecting(executed.writes().keys);
                 if (executes == null || executes.isEmpty())
                 {
-                    // TODO (desirable, performance): Mark no-ops in CFK so we can notify later transactions immediately
                     logger.trace("{}: applying no-op", txnId);
                     safeCommand.applied(safeStore);
                     safeStore.notifyListeners(safeCommand, command);
                 }
                 else
                 {
-                    // TODO (now): we should set applying within apply to avoid applying multiple times
                     safeCommand.applying(safeStore);
                     safeStore.notifyListeners(safeCommand, command);
                     logger.trace("{}: applying", command.txnId());
diff --git a/accord-core/src/main/java/accord/local/CommonAttributes.java b/accord-core/src/main/java/accord/local/CommonAttributes.java
@@ -25,6 +25,8 @@
 import accord.primitives.TxnId;
 import accord.utils.Invariants;
 
+import static accord.primitives.Status.Durability.NotDurable;
+
 public interface CommonAttributes
 {
     TxnId txnId();
@@ -51,6 +53,7 @@ public Mutable(TxnId txnId)
         {
             this.txnId = txnId;
             this.participants = StoreParticipants.empty(txnId);
+            this.durability = NotDurable;
         }
 
         public Mutable(CommonAttributes attributes)
diff --git a/accord-core/src/main/java/accord/primitives/Timestamp.java b/accord-core/src/main/java/accord/primitives/Timestamp.java
@@ -209,19 +209,28 @@ public Timestamp next()
     public int compareTo(@Nonnull Timestamp that)
     {
         if (this == that) return 0;
-        int c = Long.compareUnsigned(this.msb, that.msb);
-        if (c == 0) c = Long.compare(lowHlc(this.lsb), lowHlc(that.lsb));
-        if (c == 0) c = Long.compare(this.lsb & IDENTITY_FLAGS, that.lsb & IDENTITY_FLAGS);
+        int c = compareMsb(this.msb, that.msb);
+        if (c == 0) c = compareLsb(this.lsb, that.lsb);
         if (c == 0) c = this.node.compareTo(that.node);
         return c;
     }
 
+    public static int compareMsb(long msbA, long msbB)
+    {
+        return Long.compareUnsigned(msbA, msbB);
+    }
+
+    public static int compareLsb(long lsbA, long lsbB)
+    {
+        int c = Long.compare(lowHlc(lsbA), lowHlc(lsbB));
+        return c != 0 ? c : Long.compare(lsbA & IDENTITY_FLAGS, lsbB & IDENTITY_FLAGS);
+    }
+
     public int compareToWithoutEpoch(@Nonnull Timestamp that)
     {
         if (this == that) return 0;
         int c = Long.compare(highHlc(this.msb), highHlc(that.msb));
-        if (c == 0) c = Long.compare(lowHlc(this.lsb), lowHlc(that.lsb));
-        if (c == 0) c = Long.compare(this.lsb & IDENTITY_FLAGS, that.lsb & IDENTITY_FLAGS);
+        if (c == 0) c = compareLsb(this.lsb, that.lsb);
         if (c == 0) c = this.node.compareTo(that.node);
         return c;
     }
diff --git a/accord-core/src/main/java/accord/topology/TopologyManager.java b/accord-core/src/main/java/accord/topology/TopologyManager.java
@@ -731,7 +731,7 @@ private Topologies withSufficientEpochsAtLeast(Unseekables<?> select, long minEp
         // An issue was found where a range was removed from a replica and min selection picked the epoch before that,
         // which caused a node to get included in the txn that actually lost the range
         // See CASSANDRA-18804
-        while (i < maxi && !select.isEmpty())
+        while (i < maxi)
         {
             EpochState epochState = snapshot.epochs[i++];
             topologies.add(epochState.global.forSelection(select, false));
diff --git a/accord-core/src/main/java/accord/utils/ArrayBuffers.java b/accord-core/src/main/java/accord/utils/ArrayBuffers.java
@@ -23,7 +23,9 @@
 import accord.primitives.Range;
 import accord.primitives.TxnId;
 
+import java.io.Closeable;
 import java.lang.reflect.Array;
+import java.util.AbstractList;
 import java.util.Arrays;
 import java.util.function.IntFunction;
 
@@ -783,4 +785,38 @@ public void realDiscard(int[] buffer, int size)
         }
     }
 
+    public static class BufferList<E> extends AbstractList<E> implements Closeable
+    {
+        private static final Object[] EMPTY = new Object[0];
+        private Object[] buffer = EMPTY;
+        private int size;
+
+        @Override
+        public E get(int index)
+        {
+            return (E) buffer[index];
+        }
+
+        @Override
+        public int size()
+        {
+            return size;
+        }
+
+        @Override
+        public boolean add(E e)
+        {
+            if (size == buffer.length)
+                buffer = cachedAny().resize(buffer, size, Math.max(8, size * 2));
+            buffer[size++] = e;
+            return true;
+        }
+
+        public void close()
+        {
+            if (buffer == null) return;
+            cachedAny().forceDiscard(buffer, size);
+            buffer = null;
+        }
+    }
 }
diff --git a/accord-core/src/main/java/accord/utils/LogGroupTimers.java b/accord-core/src/main/java/accord/utils/LogGroupTimers.java
@@ -95,7 +95,7 @@ public Scheduling(Scheduler scheduler, Function<T, Runnable> taskFactory, long s
 
         public void ensureScheduled(long now)
         {
-            now = Math.max(lastNow, now);
+            lastNow = now = Math.max(lastNow, now);
             T next = peekIfSoonerThan(now + preciseDelayThreshold);
             long runAt;
             if (next == null)
diff --git a/accord-core/src/test/java/accord/impl/LocalListenersTest.java b/accord-core/src/test/java/accord/impl/LocalListenersTest.java
@@ -430,18 +430,20 @@ static class TestCommand extends Command
         final TxnId txnId;
         final SaveStatus saveStatus;
         final Durability durability;
+        final StoreParticipants participants;
 
         TestCommand(TxnId txnId, SaveStatus saveStatus, Durability durability)
         {
             this.txnId = txnId;
             this.saveStatus = saveStatus;
             this.durability = durability;
+            this.participants = StoreParticipants.empty(txnId);
         }
 
         @Override
         public StoreParticipants participants()
         {
-            return null;
+            return participants;
         }
 
         @Override
diff --git a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
@@ -131,7 +131,7 @@ void noConflictWitnessTest()
             Assertions.assertNull(command.executeAt());
         }
         SafeCommandStore safeStore = commands.beginOperation(PreLoadContext.contextFor(txnId, keys.toParticipants()));
-        StoreParticipants participants = StoreParticipants.update(safeStore, keys.toParticipants(), txnId.epoch(), txnId, txnId.epoch());
+        StoreParticipants participants = StoreParticipants.update(safeStore, ROUTE, txnId.epoch(), txnId, txnId.epoch());
         SafeCommand  safeCommand = safeStore.get(txnId, participants);
         Commands.preaccept(safeStore, safeCommand, participants, txnId, txnId.epoch(), txn.slice(FULL_RANGES, true), ROUTE);
         Command command = safeStore.get(txnId).current();
@@ -161,7 +161,7 @@ void supersedingEpochWitnessTest() throws ExecutionException {
         ((TestableConfigurationService)node.configService()).reportTopology(TopologyUtils.withEpoch(support.local.get(), 2));
         Timestamp expectedTimestamp = Timestamp.fromValues(2, 110, ID1);
         getUninterruptibly(commands.execute(context, (Consumer<? super SafeCommandStore>) safeStore -> {
-            StoreParticipants participants = StoreParticipants.update(safeStore, keys.toParticipants(), txnId.epoch(), txnId, 2);
+            StoreParticipants participants = StoreParticipants.update(safeStore, ROUTE, txnId.epoch(), txnId, 2);
             Commands.preaccept(safeStore, safeStore.get(txnId, participants), participants, txnId, txnId.epoch(), txn.slice(FULL_RANGES, true), ROUTE);
         }));
         commands.execute(PreLoadContext.contextFor(txnId, txn.keys().toParticipants()), safeStore -> {
diff --git a/accord-core/src/test/java/accord/primitives/KeyDepsTest.java b/accord-core/src/test/java/accord/primitives/KeyDepsTest.java
diff --git a/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java b/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java

Original file line number	Diff line number	Diff line change
`@@ -182,7 +182,7 @@ private void createSyncPoint()`
`182`	`182`	`{`
`183`	`183`	`AsyncSyncPoint async = syncPoint.apply(node, route);`
`184`	`184`	`coordinateSyncPoint = async.async;`
`185`		`- if (!barrierType.global)`
	`185`	`+ if (barrierType.async)`
`186`	`186`	`{`
`187`	`187`	`Invariants.checkState(barrierType.async);`
`188`	`188`	`TxnId txnId = async.txnId;`
Original file line number	Diff line number	Diff line change
`@@ -37,8 +37,13 @@ public class ErasedSafeCommand extends SafeCommand`
`37`	`37`	`public ErasedSafeCommand(TxnId txnId, SaveStatus saveStatus)`
`38`	`38`	`{`
`39`	`39`	`super(txnId);`
	`40`	`+ this.erased = erased(txnId, saveStatus);`
	`41`	`+ }`
	`42`	`+`
	`43`	`+ public static Command erased(TxnId txnId, SaveStatus saveStatus)`
	`44`	`+ {`
`40`	`45`	`Invariants.checkArgument(saveStatus.compareTo(Erased) >= 0);`
`41`		`- this.erased = new Command.Truncated(txnId, saveStatus, saveStatus == ErasedOrVestigial ? NotDurable : UniversalOrInvalidated, StoreParticipants.empty(txnId), null, null, null);`
	`46`	`+ return new Command.Truncated(txnId, saveStatus, saveStatus == ErasedOrVestigial ? NotDurable : UniversalOrInvalidated, StoreParticipants.empty(txnId), null, null, null);`
`42`	`47`	`}`
`43`	`48`
`44`	`49`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -197,18 +197,18 @@ private AbstractCommand(TxnId txnId, SaveStatus status, Durability durability, @`
`197`	`197`	`{`
`198`	`198`	`this.txnId = txnId;`
`199`	`199`	`this.status = validateCommandClass(txnId, status, getClass());`
`200`		`- this.durability = durability;`
`201`		`- this.participants = participants;`
`202`		`- this.promised = promised;`
	`200`	`+ this.durability = Invariants.nonNull(durability);`
	`201`	`+ this.participants = Invariants.nonNull(participants);`
	`202`	`+ this.promised = Invariants.nonNull(promised);`
`203`	`203`	`}`
`204`	`204`
`205`	`205`	`private AbstractCommand(CommonAttributes common, SaveStatus status, Ballot promised)`
`206`	`206`	`{`
`207`	`207`	`this.txnId = common.txnId();`
`208`	`208`	`this.status = validateCommandClass(txnId, status, getClass());`
`209`		`- this.durability = common.durability();`
`210`		`- this.participants = common.participants();`
`211`		`- this.promised = promised;`
	`209`	`+ this.durability = Invariants.nonNull(common.durability());`
	`210`	`+ this.participants = Invariants.nonNull(common.participants());`
	`211`	`+ this.promised = Invariants.nonNull(promised);`
`212`	`212`	`}`
`213`	`213`
`214`	`214`	`@Override`
`@@ -1020,6 +1020,8 @@ private Committed(CommonAttributes common, SaveStatus status, Timestamp executeA`
`1020`	`1020`	`{`
`1021`	`1021`	`super(common, status, promised, executeAt, accepted);`
`1022`	`1022`	`this.waitingOn = waitingOn;`
	`1023`	`+ Invariants.nonNull(common.participants());`
	`1024`	`+ Invariants.nonNull(common.route());`
`1023`	`1025`	`Invariants.checkState(common.route().kind().isFullRoute(), "Expected a full route but given %s", common.route().kind());`
`1024`	`1026`	`}`
`1025`	`1027`
Original file line number	Diff line number	Diff line change
`@@ -605,14 +605,12 @@ public static boolean maybeExecute(SafeCommandStore safeStore, SafeCommand safeC`
`605`	`605`	`.intersecting(executed.writes().keys);`
`606`	`606`	`if (executes == null \|\| executes.isEmpty())`
`607`	`607`	`{`
`608`		`- // TODO (desirable, performance): Mark no-ops in CFK so we can notify later transactions immediately`
`609`	`608`	`logger.trace("{}: applying no-op", txnId);`
`610`	`609`	`safeCommand.applied(safeStore);`
`611`	`610`	`safeStore.notifyListeners(safeCommand, command);`
`612`	`611`	`}`
`613`	`612`	`else`
`614`	`613`	`{`
`615`		`- // TODO (now): we should set applying within apply to avoid applying multiple times`
`616`	`614`	`safeCommand.applying(safeStore);`
`617`	`615`	`safeStore.notifyListeners(safeCommand, command);`
`618`	`616`	`logger.trace("{}: applying", command.txnId());`
Original file line number	Diff line number	Diff line change
`@@ -25,6 +25,8 @@`
`25`	`25`	`import accord.primitives.TxnId;`
`26`	`26`	`import accord.utils.Invariants;`
`27`	`27`
	`28`	`+import static accord.primitives.Status.Durability.NotDurable;`
	`29`	`+`
`28`	`30`	`public interface CommonAttributes`
`29`	`31`	`{`
`30`	`32`	`TxnId txnId();`
`@@ -51,6 +53,7 @@ public Mutable(TxnId txnId)`
`51`	`53`	`{`
`52`	`54`	`this.txnId = txnId;`
`53`	`55`	`this.participants = StoreParticipants.empty(txnId);`
	`56`	`+ this.durability = NotDurable;`
`54`	`57`	`}`
`55`	`58`
`56`	`59`	`public Mutable(CommonAttributes attributes)`
Original file line number	Diff line number	Diff line change
`@@ -731,7 +731,7 @@ private Topologies withSufficientEpochsAtLeast(Unseekables<?> select, long minEp`
`731`	`731`	`// An issue was found where a range was removed from a replica and min selection picked the epoch before that,`
`732`	`732`	`// which caused a node to get included in the txn that actually lost the range`
`733`	`733`	`// See CASSANDRA-18804`
`734`		`- while (i < maxi && !select.isEmpty())`
	`734`	`+ while (i < maxi)`
`735`	`735`	`{`
`736`	`736`	`EpochState epochState = snapshot.epochs[i++];`
`737`	`737`	`topologies.add(epochState.global.forSelection(select, false));`