From ec7c7c9fce45437778e53e000b37990bcc2b7e78 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Tue, 19 Nov 2024 20:04:11 +0000 Subject: [PATCH] Accord: Fix unit tests and improve burn test stability patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20113 --- modules/accord | 2 +- .../service/accord/AccordCommandStore.java | 22 ++- .../service/accord/AccordExecutor.java | 3 +- .../accord/AccordSafeCommandStore.java | 161 +----------------- .../service/accord/AccordService.java | 6 +- .../service/accord/CommandsForRanges.java | 7 +- .../accord/interop/AccordInteropAdapter.java | 13 +- .../accord/interop/AccordInteropApply.java | 10 +- .../accord/interop/AccordInteropPersist.java | 5 +- .../accord/serializers/AwaitSerializer.java | 11 +- .../serializers/ReadDataSerializers.java | 14 +- .../CompactionAccordIteratorsTest.java | 15 +- .../service/accord/AccordTaskTest.java | 2 +- .../CommandsForKeySerializerTest.java | 13 +- .../cassandra/utils/AccordGenerators.java | 16 +- 15 files changed, 98 insertions(+), 202 deletions(-) diff --git a/modules/accord b/modules/accord index a271897790..61f4ebd21f 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit a271897790aa3816c3dea2125b1e374b091bc090 +Subproject commit 61f4ebd21ff34b0807081d9df442236d8c334b52 diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index d3c59fbd7f..f0b4eb12b5 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -38,6 +38,7 @@ import accord.api.DataStore; import accord.api.LocalListeners; import accord.api.ProgressLog; import accord.api.RoutingKey; +import accord.impl.AbstractSafeCommandStore.CommandStoreCaches; import accord.impl.TimestampsForKey; import accord.local.Cleanup; import accord.local.Command; @@ -119,7 +120,7 @@ public class AccordCommandStore extends CommandStore } } - public static final class ExclusiveCaches extends Caches implements AutoCloseable + public static final class ExclusiveCaches extends Caches implements CommandStoreCaches { private final Lock lock; @@ -129,6 +130,25 @@ public class AccordCommandStore extends CommandStore this.lock = lock; } + + @Override + public AccordSafeCommand acquireIfLoaded(TxnId txnId) + { + return commands().acquireIfLoaded(txnId); + } + + @Override + public AccordSafeCommandsForKey acquireIfLoaded(RoutingKey key) + { + return commandsForKeys().acquireIfLoaded(key); + } + + @Override + public AccordSafeTimestampsForKey acquireTfkIfLoaded(RoutingKey key) + { + return timestampsForKeys().acquireIfLoaded(key); + } + @Override public void close() { diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java index ce778dd76a..cc6e96364d 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java @@ -561,7 +561,8 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo private void onLoadedExclusive(AccordCacheEntry loaded, V value, Throwable fail, boolean isForRange) { --activeLoads; - --activeRangeLoads; + if (isForRange) + --activeRangeLoads; if (loaded.status() != EVICTED) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index 8164214298..c582e15f9e 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -18,9 +18,7 @@ package org.apache.cassandra.service.accord; -import java.util.ArrayList; import java.util.Collections; -import java.util.List; import java.util.Map; import java.util.NavigableMap; import java.util.Set; @@ -31,26 +29,18 @@ import com.google.common.annotations.VisibleForTesting; import accord.api.Agent; import accord.api.DataStore; -import accord.api.Key; import accord.api.ProgressLog; import accord.api.RoutingKey; +import accord.impl.AbstractSafeCommandStore; import accord.impl.CommandsSummary; -import accord.impl.SafeTimestampsForKey; import accord.local.CommandStores; import accord.local.CommandStores.RangesForEpoch; -import accord.local.KeyHistory; import accord.local.NodeCommandStoreService; -import accord.local.PreLoadContext; import accord.local.RedundantBefore; -import accord.local.SafeCommandStore; import accord.local.cfk.CommandsForKey; import accord.primitives.AbstractKeys; -import accord.primitives.AbstractUnseekableKeys; -import accord.primitives.Participants; import accord.primitives.Ranges; -import accord.primitives.Routable; import accord.primitives.Routables; -import accord.primitives.RoutingKeys; import accord.primitives.Timestamp; import accord.primitives.Txn; import accord.primitives.TxnId; @@ -58,14 +48,11 @@ import accord.primitives.Unseekables; import accord.utils.Invariants; import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches; -import static accord.local.KeyHistory.TIMESTAMPS; -import static accord.utils.Invariants.illegalArgument; import static accord.utils.Invariants.illegalState; -public class AccordSafeCommandStore extends SafeCommandStore +public class AccordSafeCommandStore extends AbstractSafeCommandStore { final AccordTask task; - final PreLoadContext context; private final @Nullable CommandsForRanges commandsForRanges; private final AccordCommandStore commandStore; private RangesForEpoch ranges; @@ -75,7 +62,7 @@ public class AccordSafeCommandStore extends SafeCommandStore @Nullable CommandsForRanges commandsForRanges, AccordCommandStore commandStore) { - this.context = task.preLoadContext(); + super(task.preLoadContext()); this.task = task; this.commandsForRanges = commandsForRanges; this.commandStore = commandStore; @@ -91,91 +78,6 @@ public class AccordSafeCommandStore extends SafeCommandStore return new AccordSafeCommandStore(operation, commandsForRanges, commandStore); } - @Override - public PreLoadContext canExecute(PreLoadContext with) - { - if (with.isEmpty()) return with; - if (with.keys().domain() == Routable.Domain.Range) - return with.isSubsetOf(this.context) ? with : null; - - if (!context().keyHistory().satisfies(with.keyHistory())) - return null; - - try (ExclusiveCaches caches = commandStore.tryLockCaches()) - { - if (caches == null) - return with.isSubsetOf(this.context) ? with : null; - - for (TxnId txnId : with.txnIds()) - { - if (null != getInternal(txnId)) - continue; - - AccordSafeCommand safeCommand = caches.commands().acquireIfLoaded(txnId); - if (safeCommand == null) - return null; - - add(safeCommand, caches); - } - - KeyHistory keyHistory = with.keyHistory(); - if (keyHistory == KeyHistory.NONE) - return with; - - List unavailable = null; - AbstractUnseekableKeys keys = (AbstractUnseekableKeys) context.keys(); - if (keys.size() == 0) - return with; - - for (int i = 0 ; i < keys.size() ; ++i) - { - RoutingKey key = keys.get(i); - if (keyHistory == TIMESTAMPS) - { - if (null != timestampsForKeyInternal(key)) - continue; // already in working set - - AccordSafeTimestampsForKey safeTfk = caches.timestampsForKeys().acquireIfLoaded(key); - if (safeTfk != null) - { - add(safeTfk, caches); - continue; - } - } - else - { - if (null != getInternal(key)) - continue; // already in working set - - AccordSafeCommandsForKey safeCfk = caches.commandsForKeys().acquireIfLoaded(key); - if (safeCfk != null) - { - add(safeCfk, caches); - continue; - } - } - if (unavailable == null) - unavailable = new ArrayList<>(); - unavailable.add(key); - } - - if (unavailable == null) - return with; - - if (unavailable.size() == keys.size()) - return null; - - Participants available = keys.without(RoutingKeys.ofSortedUnique(unavailable)); - return PreLoadContext.contextFor(context.primaryTxnId(), context.additionalTxnId(), available, keyHistory); - } - } - - @Override - public PreLoadContext context() - { - return context; - } - @VisibleForTesting public Set commandsForKeysKeys() { @@ -194,22 +96,12 @@ public class AccordSafeCommandStore extends SafeCommandStore } @Override - protected AccordSafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId) + protected ExclusiveCaches tryGetCaches() { - try (ExclusiveCaches caches = commandStore.tryLockCaches()) - { - if (caches == null) - return null; - - AccordSafeCommand command = caches.commands().acquireIfLoaded(txnId); - if (command == null) - return null; - - return add(command, caches); - } + return commandStore.tryLockCaches(); } - private AccordSafeCommand add(AccordSafeCommand safeCommand, ExclusiveCaches caches) + protected AccordSafeCommand add(AccordSafeCommand safeCommand, ExclusiveCaches caches) { Object check = task.ensureCommands().putIfAbsent(safeCommand.txnId(), safeCommand); if (check == null) @@ -224,7 +116,7 @@ public class AccordSafeCommandStore extends SafeCommandStore } } - private AccordSafeCommandsForKey add(AccordSafeCommandsForKey safeCfk, ExclusiveCaches caches) + protected AccordSafeCommandsForKey add(AccordSafeCommandsForKey safeCfk, ExclusiveCaches caches) { Object check = task.ensureCommandsForKey().putIfAbsent(safeCfk.key(), safeCfk); if (check == null) @@ -239,7 +131,7 @@ public class AccordSafeCommandStore extends SafeCommandStore } } - private AccordSafeTimestampsForKey add(AccordSafeTimestampsForKey safeTfk, ExclusiveCaches caches) + protected AccordSafeTimestampsForKey add(AccordSafeTimestampsForKey safeTfk, ExclusiveCaches caches) { Object check = task.ensureTimestampsForKey().putIfAbsent(safeTfk.key(), safeTfk); if (check == null) @@ -263,42 +155,7 @@ public class AccordSafeCommandStore extends SafeCommandStore return commandsForKey.get(key); } - @Override - protected AccordSafeCommandsForKey ifLoadedInternal(RoutingKey key) - { - try (ExclusiveCaches caches = commandStore.tryLockCaches()) - { - if (caches == null) - return null; - - AccordSafeCommandsForKey safeCfk = caches.commandsForKeys().acquireIfLoaded(key); - if (safeCfk == null) - return null; - - Object check = task.ensureCommandsForKey().putIfAbsent(safeCfk.key(), safeCfk); - if (check == null) - { - safeCfk.preExecute(); - return safeCfk; - } - else - { - caches.commandsForKeys().release(safeCfk, task); - throw illegalState("Attempted to take a duplicate reference to CFK for %s", key); - } - } - } - - @Override - public SafeTimestampsForKey timestampsForKey(RoutingKey key) - { - AccordSafeTimestampsForKey safeTfk = timestampsForKeyInternal(key); - if (safeTfk == null) - throw illegalArgument("%s not referenced in %s", key, context); - return safeTfk; - } - - private AccordSafeTimestampsForKey timestampsForKeyInternal(RoutingKey key) + protected AccordSafeTimestampsForKey timestampsForKeyInternal(RoutingKey key) { Map timestampsForKey = task.timestampsForKey(); if (timestampsForKey == null) diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 0e33508378..9cd8f36237 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -1037,7 +1037,7 @@ public class AccordService implements IAccordService, Shutdownable { if (state.knows(blockedBy)) continue; // need to fetch the state - if (safeStore.ifLoadedAndInitialisedAndNotErased(blockedBy) != null) + if (safeStore.ifLoadedAndInitialised(blockedBy) != null) { AsyncChain chain = populate(state, safeStore, blockedBy); if (chain != null) @@ -1077,7 +1077,7 @@ public class AccordService implements IAccordService, Shutdownable blocking = ((CommandsForKey.TxnInfo) blocking).plainTxnId(); state.keys.put(pk, blocking); if (state.txns.containsKey(blocking)) return null; - if (safeStore.ifLoadedAndInitialisedAndNotErased(blocking) != null) return populate(state, safeStore, blocking); + if (safeStore.ifLoadedAndInitialised(blocking) != null) return populate(state, safeStore, blocking); return populate(state, safeStore.commandStore(), blocking); } @@ -1437,7 +1437,7 @@ public class AccordService implements IAccordService, Shutdownable { if (tracker.recordSuccess(from) == RequestStatus.Success) { - node.configService().reportEpochRedundant(exclusiveSyncPoint.route.toRanges(), exclusiveSyncPoint.syncId.epoch()); + node.configService().reportEpochRedundant(exclusiveSyncPoint.route.toRanges(), exclusiveSyncPoint.syncId.epoch() - 1); trySuccess(exclusiveSyncPoint); } } diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java index 6cae377acd..2bae3b6c94 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java @@ -57,7 +57,7 @@ import org.apache.cassandra.index.accord.RoutesSearcher; import org.apache.cassandra.service.accord.api.AccordRoutingKey; import static accord.local.SafeCommandStore.TestDep.ANY_DEPS; -import static accord.local.SafeCommandStore.TestDep.WITH; +import static accord.local.SafeCommandStore.TestDep.WITH_OR_INVALIDATED; import static accord.local.SafeCommandStore.TestStartedAt.STARTED_BEFORE; import static accord.local.SafeCommandStore.TestStatus.ANY_STATUS; import static accord.primitives.Routables.Slice.Minimal; @@ -129,6 +129,9 @@ public class CommandsForRanges extends TreeMap route, Participants sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer callback) + public void persist(Node node, Topologies all, FullRoute fullRoute, Route sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer callback) { - // TODO (required): we aren't using sendTo - if (applyKind == Minimal && doInteropPersist(node, all, route, txnId, txn, executeAt, deps, writes, result, callback)) + if (applyKind == Minimal && doInteropPersist(node, all, sendTo, txnId, txn, executeAt, deps, writes, result, fullRoute, callback)) return; if (callback != null) callback.accept(result, null); - new PersistTxn(node, all, txnId, route, txn, executeAt, deps, writes, result) + new PersistTxn(node, all, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute) .start(Apply.FACTORY, applyKind, all, writes, result); } @@ -113,14 +112,14 @@ public class AccordInteropAdapter extends AbstractTxnAdapter return true; } - private static boolean doInteropPersist(Node node, Topologies all, FullRoute route, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer callback) + private static boolean doInteropPersist(Node node, Topologies all, Route sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute fullRoute, BiConsumer callback) { Update update = txn.update(); ConsistencyLevel consistencyLevel = update instanceof AccordUpdate ? ((AccordUpdate) update).cassandraCommitCL() : null; if (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ANY || writes.isEmpty()) return false; - new AccordInteropPersist(node, all, txnId, route, txn, executeAt, deps, writes, result, consistencyLevel, callback) + new AccordInteropPersist(node, all, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, callback) .start(AccordInteropApply.FACTORY, Minimal, all, writes, result); return true; } diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java index 1cd3a05695..44f54ab51d 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java @@ -62,14 +62,14 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL public static final Apply.Factory FACTORY = new Apply.Factory() { @Override - public Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, FullRoute route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result) + public Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, Route route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute fullRoute) { checkArgument(kind != Kind.Maximal, "Shouldn't need to send a maximal commit with interop support"); ConsistencyLevel commitCL = txn.update() instanceof AccordUpdate ? ((AccordUpdate) txn.update()).cassandraCommitCL() : null; // Any asynchronous apply option should use the regular Apply that doesn't wait for writes to complete if (commitCL == null || commitCL == ConsistencyLevel.ANY) - return Apply.FACTORY.create(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result); - return new AccordInteropApply(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result); + return Apply.FACTORY.create(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result, fullRoute); + return new AccordInteropApply(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result, fullRoute); } }; @@ -91,9 +91,9 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL super(kind, txnId, route, waitForEpoch, executeAt, deps, txn, fullRoute, writes, result); } - private AccordInteropApply(Kind kind, Id to, Topologies participates, TxnId txnId, FullRoute route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result) + private AccordInteropApply(Kind kind, Id to, Topologies participates, TxnId txnId, Route route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute fullRoute) { - super(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result); + super(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result, fullRoute); } @Override diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java index a0fd42229f..342dc40140 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java @@ -30,6 +30,7 @@ import accord.local.Node; import accord.messages.Apply; import accord.primitives.Deps; import accord.primitives.FullRoute; +import accord.primitives.Route; import accord.primitives.Timestamp; import accord.primitives.Txn; import accord.primitives.TxnId; @@ -91,9 +92,9 @@ public class AccordInteropPersist extends Persist private final ConsistencyLevel consistencyLevel; private CallbackHolder callback; - public AccordInteropPersist(Node node, Topologies topologies, TxnId txnId, FullRoute route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, ConsistencyLevel consistencyLevel, BiConsumer clientCallback) + public AccordInteropPersist(Node node, Topologies topologies, TxnId txnId, Route sendTo, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute fullRoute, ConsistencyLevel consistencyLevel, BiConsumer clientCallback) { - super(node, topologies, txnId, route, txn, executeAt, deps, writes, result); + super(node, topologies, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute); Invariants.checkArgument(consistencyLevel == ConsistencyLevel.QUORUM || consistencyLevel == ConsistencyLevel.ALL || consistencyLevel == ConsistencyLevel.SERIAL || consistencyLevel == ConsistencyLevel.ONE); this.consistencyLevel = consistencyLevel; registerClientCallback(result, clientCallback); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java index 54b25a1c87..0006a068ec 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java @@ -45,7 +45,8 @@ public class AwaitSerializer CommandSerializers.txnId.serialize(await.txnId, out, version); KeySerializers.participants.serialize(await.scope, out, version); out.writeByte(await.blockedUntil.ordinal()); - out.writeUnsignedVInt(await.awaitEpoch - await.txnId.epoch()); + out.writeUnsignedVInt(await.maxAwaitEpoch - await.txnId.epoch()); + out.writeUnsignedVInt(await.maxAwaitEpoch - await.minAwaitEpoch); out.writeUnsignedVInt32(await.callbackId + 1); Invariants.checkState(await.callbackId >= -1); } @@ -56,10 +57,11 @@ public class AwaitSerializer TxnId txnId = CommandSerializers.txnId.deserialize(in, version); Participants scope = KeySerializers.participants.deserialize(in, version); BlockedUntil blockedUntil = BlockedUntil.forOrdinal(in.readByte()); - long awaitEpoch = in.readUnsignedVInt() + txnId.epoch(); + long maxAwaitEpoch = in.readUnsignedVInt() + txnId.epoch(); + long minAwaitEpoch = maxAwaitEpoch - in.readUnsignedVInt(); int callbackId = in.readUnsignedVInt32() - 1; Invariants.checkState(callbackId >= -1); - return Await.SerializerSupport.create(txnId, scope, blockedUntil, awaitEpoch, callbackId); + return Await.SerializerSupport.create(txnId, scope, blockedUntil, minAwaitEpoch, maxAwaitEpoch, callbackId); } @Override @@ -68,7 +70,8 @@ public class AwaitSerializer return CommandSerializers.txnId.serializedSize(await.txnId, version) + KeySerializers.participants.serializedSize(await.scope, version) + TypeSizes.BYTE_SIZE - + VIntCoding.computeUnsignedVIntSize(await.awaitEpoch - await.txnId.epoch()) + + VIntCoding.computeUnsignedVIntSize(await.maxAwaitEpoch - await.txnId.epoch()) + + VIntCoding.computeUnsignedVIntSize(await.maxAwaitEpoch - await.minAwaitEpoch) + VIntCoding.computeUnsignedVIntSize(await.callbackId + 1); } }; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java index 0c86ee7747..947b166619 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java @@ -80,6 +80,7 @@ public class ReadDataSerializers { CommandSerializers.txnId.serialize(msg.txnId, out, version); KeySerializers.participants.serialize(msg.readScope, out, version); + out.writeUnsignedVInt(msg.minEpoch()); CommandSerializers.timestamp.serialize(msg.executeAt, out, version); KeySerializers.fullRoute.serialize(msg.route, out, version); CommandSerializers.partialTxn.serialize(msg.txn, out, version); @@ -93,6 +94,7 @@ public class ReadDataSerializers return ApplyThenWaitUntilApplied.SerializerSupport.create( CommandSerializers.txnId.deserialize(in, version), KeySerializers.participants.deserialize(in, version), + in.readUnsignedVInt(), CommandSerializers.timestamp.deserialize(in, version), KeySerializers.fullRoute.deserialize(in, version), CommandSerializers.partialTxn.deserialize(in, version), @@ -106,6 +108,7 @@ public class ReadDataSerializers { return CommandSerializers.txnId.serializedSize(msg.txnId, version) + KeySerializers.participants.serializedSize(msg.readScope, version) + + TypeSizes.sizeofUnsignedVInt(msg.minEpoch()) + CommandSerializers.timestamp.serializedSize(msg.executeAt, version) + KeySerializers.fullRoute.serializedSize(msg.route, version) + CommandSerializers.partialTxn.serializedSize(msg.txn, version) @@ -279,7 +282,8 @@ public class ReadDataSerializers { CommandSerializers.txnId.serialize(waitUntilApplied.txnId, out, version); KeySerializers.participants.serialize(waitUntilApplied.readScope, out, version); - out.writeUnsignedVInt(waitUntilApplied.executeAtEpoch); + out.writeUnsignedVInt(waitUntilApplied.minEpoch()); + out.writeUnsignedVInt(waitUntilApplied.executeAtEpoch - waitUntilApplied.minEpoch()); } @Override @@ -287,8 +291,9 @@ public class ReadDataSerializers { TxnId txnId = CommandSerializers.txnId.deserialize(in, version); Participants readScope = KeySerializers.participants.deserialize(in, version); - long executeAtEpoch = in.readUnsignedVInt(); - return WaitUntilApplied.SerializerSupport.create(txnId, readScope, executeAtEpoch); + long minEpoch = in.readUnsignedVInt(); + long executeAtEpoch = minEpoch + in.readUnsignedVInt(); + return WaitUntilApplied.SerializerSupport.create(txnId, readScope, minEpoch, executeAtEpoch); } @Override @@ -296,7 +301,8 @@ public class ReadDataSerializers { return CommandSerializers.txnId.serializedSize(waitUntilApplied.txnId, version) + KeySerializers.participants.serializedSize(waitUntilApplied.readScope, version) - + TypeSizes.sizeofUnsignedVInt(waitUntilApplied.executeAtEpoch); + + TypeSizes.sizeofUnsignedVInt(waitUntilApplied.minEpoch()) + + TypeSizes.sizeofUnsignedVInt(waitUntilApplied.executeAtEpoch - waitUntilApplied.minEpoch()); } }; } diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index 6bf6849e42..fbcfee12f3 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java @@ -207,8 +207,8 @@ public class CompactionAccordIteratorsTest testAccordCommandsPurger(redundantBefore(GT_TXN_ID), durableBefore(UNIVERSAL), expectAccordCommandsErase()); // Universally durable but not locally; we're stale, but shouldn't erase testAccordCommandsPurger(redundantBefore(LT_TXN_ID), durableBefore(UNIVERSAL), expectAccordCommandsNoChange()); - // With redundantBefore at the txnId there should be no change because it is < not <= - testAccordCommandsPurger(redundantBefore(TXN_ID), durableBefore(MAJORITY), expectAccordCommandsNoChange()); + // redundantBefore will be > txnId as must be exclusive sync point, so will truncate + testAccordCommandsPurger(redundantBefore(TXN_ID), durableBefore(MAJORITY), expectAccordCommandsTruncated()); testAccordCommandsPurger(redundantBefore(LT_TXN_ID), durableBefore(MAJORITY), expectAccordCommandsNoChange()); // Durable at a majority can be truncated with minimal data preserved, it must be redundant for this to occur testAccordCommandsPurger(redundantBefore(GT_TXN_ID), durableBefore(MAJORITY), expectAccordCommandsTruncated()); @@ -249,10 +249,11 @@ public class CompactionAccordIteratorsTest testAccordCommandsForKeyPurger(null, expectedAccordCommandsForKeyNoChange()); testAccordTimestampsForKeyPurger(redundantBefore(LT_TXN_ID), expectedAccordTimestampsForKeyNoChange()); testAccordCommandsForKeyPurger(redundantBefore(LT_TXN_ID), expectedAccordCommandsForKeyNoChange()); - testAccordTimestampsForKeyPurger(redundantBefore(TXN_ID), expectedAccordTimestampsForKeyNoChange()); - testAccordCommandsForKeyPurger(redundantBefore(TXN_ID), expectedAccordCommandsForKeyNoChange()); - testAccordTimestampsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordTimestampsForKeyEraseOne()); - testAccordCommandsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordCommandsForKeyEraseOne()); + // will erase one more than expected as converted to ExclusiveSyncPoint id which is > base id + testAccordTimestampsForKeyPurger(redundantBefore(TXN_ID), expectedAccordTimestampsForKeyEraseOne()); + testAccordCommandsForKeyPurger(redundantBefore(TXN_ID), expectedAccordCommandsForKeyEraseOne()); + testAccordTimestampsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordTimestampsForKeyEraseAll()); + testAccordCommandsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordCommandsForKeyEraseAll()); testAccordTimestampsForKeyPurger(redundantBefore(GT_SECOND_TXN_ID), expectedAccordTimestampsForKeyEraseAll()); testAccordCommandsForKeyPurger(redundantBefore(GT_SECOND_TXN_ID), expectedAccordCommandsForKeyEraseAll()); } @@ -399,7 +400,7 @@ public class CompactionAccordIteratorsTest private static RedundantBefore redundantBefore(TxnId txnId) { Ranges ranges = AccordTestUtils.fullRange(AccordTestUtils.keys(table, 42)); - txnId = txnId.as(Kind.Read, Range); + txnId = txnId.as(Kind.ExclusiveSyncPoint, Range); return RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, txnId, txnId, txnId, txnId, LT_TXN_ID.as(Range)); } diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java index bc476baab5..0207b14d20 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java @@ -133,7 +133,7 @@ public class AccordTaskTest // TODO review: This change to `ifInitialized` was done in a lot of places and it doesn't preserve this property // I fixed this reference to point to `ifLoadedAndInitialised` and but didn't update other places Assert.assertNull(instance.ifInitialised(txnId)); - Assert.assertNull(instance.ifLoadedAndInitialisedAndNotErased(txnId)); + Assert.assertNull(instance.ifLoadedAndInitialised(txnId)); })); UntypedResultSet result = AccordKeyspace.loadCommandRow(commandStore, txnId); diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index 052736200c..e863661982 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -23,6 +23,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; +import java.util.EnumSet; import java.util.List; import java.util.Random; import java.util.Set; @@ -202,6 +203,8 @@ public class CommandsForKeySerializerTest case Erased: case ErasedOrVestigial: + return Command.Truncated.erased(txnId, Status.Durability.UniversalOrInvalidated, StoreParticipants.empty(txnId)); + case Invalidated: return Command.SerializerSupport.invalidated(txnId); } @@ -431,7 +434,8 @@ public class CommandsForKeySerializerTest } } - Choices saveStatusChoices = Choices.uniform(SaveStatus.values()); + // TODO (expected): we currently don't explore TruncatedApply statuses because we don't transition through all phases and therefore don't adopt the Applied status + Choices saveStatusChoices = Choices.uniform(EnumSet.complementOf(EnumSet.of(SaveStatus.TruncatedApply, SaveStatus.TruncatedApplyWithOutcome, SaveStatus.TruncatedApplyWithDeps)).toArray(SaveStatus[]::new)); Supplier saveStatusSupplier = () -> { SaveStatus result = saveStatusChoices.choose(source); while (result == SaveStatus.TruncatedApplyWithDeps) // not a real save status @@ -471,7 +475,8 @@ public class CommandsForKeySerializerTest while (commands.size() > 0) { int next = source.nextInt(commands.size()); - cfk = cfk.update(commands.get(next)).cfk(); + Command command = commands.get(next); + cfk = cfk.update(command).cfk(); commands.set(next, commands.get(commands.size() - 1)); commands.remove(commands.size() - 1); } @@ -486,7 +491,7 @@ public class CommandsForKeySerializerTest } TxnInfo info = cfk.get(i); InternalStatus expectStatus = InternalStatus.from(cmd.saveStatus); - if (expectStatus == null) expectStatus = InternalStatus.TRANSITIVELY_KNOWN; + if (expectStatus == null) expectStatus = InternalStatus.TRANSITIVE; if (expectStatus.hasExecuteAtOrDeps) Assert.assertEquals(cmd.executeAt, info.executeAt); Assert.assertEquals(expectStatus, info.status()); @@ -545,7 +550,7 @@ public class CommandsForKeySerializerTest { int idx = Arrays.binarySearch(ids, u.txnId); if (idx < 0) - missing.add(TxnInfo.create(u.txnId, InternalStatus.TRANSITIVELY_KNOWN, true, u.txnId, Ballot.ZERO)); + missing.add(TxnInfo.create(u.txnId, InternalStatus.TRANSITIVE, true, u.txnId, Ballot.ZERO)); } if (!missing.isEmpty()) { diff --git a/test/unit/org/apache/cassandra/utils/AccordGenerators.java b/test/unit/org/apache/cassandra/utils/AccordGenerators.java index 99097c064c..e4e08ac180 100644 --- a/test/unit/org/apache/cassandra/utils/AccordGenerators.java +++ b/test/unit/org/apache/cassandra/utils/AccordGenerators.java @@ -443,7 +443,7 @@ public class AccordGenerators public static Gen redundantBeforeEntry(IPartitioner partitioner) { - return redundantBeforeEntry(Gens.bools().all(), range(partitioner), AccordGens.txnIds(Gens.pick(Txn.Kind.SyncPoint, Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range)); + return redundantBeforeEntry(Gens.bools().all(), range(partitioner), AccordGens.txnIds(Gens.pick(Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range)); } public static Gen redundantBeforeEntry(Gen emptyGen, Gen rangeGen, Gen txnIdGen) @@ -451,11 +451,11 @@ public class AccordGenerators return rs -> { Range range = rangeGen.next(rs); TxnId locallyWitnessedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range - TxnId locallyAppliedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range - TxnId locallyDecidedAndAppliedOrInvalidatedBefore = locallyAppliedOrInvalidatedBefore; - TxnId shardAppliedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range - TxnId shardOnlyAppliedOrInvalidatedBefore = shardAppliedOrInvalidatedBefore; - TxnId gcBefore = shardAppliedOrInvalidatedBefore; + TxnId locallyAppliedOrInvalidatedBefore = TxnId.nonNullOrMin(locallyWitnessedOrInvalidatedBefore, emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs)); // emptyable or range + TxnId locallyDecidedAndAppliedOrInvalidatedBefore = TxnId.nonNullOrMin(locallyAppliedOrInvalidatedBefore, emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs)); // emptyable or range + TxnId shardOnlyAppliedOrInvalidatedBefore = emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs); // emptyable or range + TxnId shardAppliedOrInvalidatedBefore = TxnId.nonNullOrMin(locallyAppliedOrInvalidatedBefore, TxnId.nonNullOrMin(shardOnlyAppliedOrInvalidatedBefore, emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs))); // emptyable or range + TxnId gcBefore = TxnId.nonNullOrMin(shardAppliedOrInvalidatedBefore, emptyGen.next(rs) ? TxnId.NONE : txnIdGen.next(rs)); // emptyable or range TxnId bootstrappedAt = txnIdGen.next(rs); Timestamp staleUntilAtLeast = emptyGen.next(rs) ? null : txnIdGen.next(rs); // nullable @@ -469,7 +469,7 @@ public class AccordGenerators public static Gen redundantBefore(IPartitioner partitioner) { Gen rangeGen = rangesArbitrary(partitioner); - Gen txnIdGen = AccordGens.txnIds(Gens.pick(Txn.Kind.SyncPoint, Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range); + Gen txnIdGen = AccordGens.txnIds(Gens.pick(Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range); BiFunction entryGen = (rs, range) -> redundantBeforeEntry(Gens.bools().all(), i -> range, txnIdGen).next(rs); return AccordGens.redundantBefore(rangeGen, entryGen); } @@ -477,7 +477,7 @@ public class AccordGenerators public static Gen durableBeforeGen(IPartitioner partitioner) { Gen rangeGen = rangesArbitrary(partitioner); - Gen txnIdGen = AccordGens.txnIds(Gens.pick(Txn.Kind.SyncPoint, Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range); + Gen txnIdGen = AccordGens.txnIds(Gens.pick(Txn.Kind.ExclusiveSyncPoint), ignore -> Routable.Domain.Range); return (rs) -> { Ranges ranges = rangeGen.next(rs);