From 49b99c93194ba455ea6df4a7c88458ce2850e48d Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Thu, 24 Jul 2025 18:35:29 +0100 Subject: [PATCH] Fix DurableBefore updates: - DurableBefore.fullyContainedIn should compare txnId Also fix: - Messages with nullable state should check if they are cancelled after reading the state but before using it - Don't snapshot during replay - CFK.updateRedundantBefore incorrectly updating mayExecute and sharing byId in some cases - Refine GC with global durability bound - Fix JournalGCTest patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20786 --- modules/accord | 2 +- .../service/accord/AccordCommandStore.java | 5 ++- .../service/accord/AccordDataStore.java | 4 ++ .../fuzz/topology/JournalGCTest.java | 37 +++++++++++++------ 4 files changed, 34 insertions(+), 14 deletions(-) diff --git a/modules/accord b/modules/accord index f40826db95..090cb49bfb 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit f40826db9571af286a689deada1613ffc872e5f8 +Subproject commit 090cb49bfb52357fee0f554d5ba0f0b012195280 diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index c6b1cea4ed..a1472a76f4 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -505,6 +505,9 @@ public class AccordCommandStore extends CommandStore @Override protected void ensureDurable(Ranges ranges, RedundantBefore onCommandStoreDurable) { + if (!CommandsForKey.reportLinearizabilityViolations()) + return; + long reportId = nextDurabilityLoggingId.incrementAndGet(); logger.debug("{} awaiting local metadata durability for {} ({})", this, ranges, reportId); executor().afterSubmittedAndConsequences(() -> { @@ -515,8 +518,6 @@ public class AccordCommandStore extends CommandStore @Override public void run() { decrement(); } } - Map saving = new TreeMap<>(); - Map excluding = new TreeMap<>(); Ready ready = new Ready(); try (ExclusiveCaches caches = lockCaches()) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordDataStore.java b/src/java/org/apache/cassandra/service/accord/AccordDataStore.java index 83e92321fb..0234184a38 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordDataStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordDataStore.java @@ -26,6 +26,7 @@ import accord.local.CommandStore; import accord.local.Node; import accord.local.RedundantBefore; import accord.local.SafeCommandStore; +import accord.local.cfk.CommandsForKey; import accord.primitives.Range; import accord.primitives.Ranges; import accord.primitives.SyncPoint; @@ -53,6 +54,9 @@ public class AccordDataStore implements DataStore */ public void ensureDurable(CommandStore commandStore, Ranges ranges, RedundantBefore reportOnSuccess) { + if (!CommandsForKey.reportLinearizabilityViolations()) + return; + logger.debug("{} awaiting local data durability of {}", commandStore, ranges); ColumnFamilyStore prev = null; for (Range range : ranges) diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java index 03c3152808..0c4f0324be 100644 --- a/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java +++ b/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java @@ -20,10 +20,12 @@ package org.apache.cassandra.fuzz.topology; import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.junit.Assert; import org.junit.Test; +import accord.primitives.TxnId; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; @@ -40,8 +42,10 @@ import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.accord.AccordKeyspace; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.service.accord.JournalKey; +import org.apache.cassandra.service.accord.TokenRange; import org.apache.cassandra.service.consensus.TransactionalMode; +import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS; import static org.apache.cassandra.harry.checker.TestHelper.withRandom; public class JournalGCTest extends FuzzTestBase @@ -104,29 +108,40 @@ public class JournalGCTest extends FuzzTestBase ((AccordService) AccordService.instance()).journal().compactor().run(); }); - Callable countDiffs = cluster.get(1).callsOnInstance(() -> { - AtomicInteger a = new AtomicInteger(); + String maximumId = cluster.get(1).callOnInstance(() -> { + AtomicReference a = new AtomicReference<>(); ((AccordService) AccordService.instance()).journal().forEach((v) -> { - if (v.type == JournalKey.Type.COMMAND_DIFF && - // Do not count syncpoints - !v.id.isSyncPoint()) + if (v.type == JournalKey.Type.COMMAND_DIFF && (a.get() == null || v.id.compareTo(a.get()) > 0)) + a.set(v.id); + }); + return a.get() == null ? "" : a.get().toString(); + }); + + Callable countDiffs = () -> cluster.get(1).applyOnInstance(maxIdStr -> { + AtomicInteger a = new AtomicInteger(); + TxnId maxId = TxnId.parse(maxIdStr); + ((AccordService) AccordService.instance()).journal().forEach((v) -> { + if (v.type == JournalKey.Type.COMMAND_DIFF && v.id.compareTo(maxId) <= 0) a.incrementAndGet(); }); return a.get(); - }); + }, maximumId); int after =-1; - for (int i = 0; i < 60; i++) + int maxCycles = 3; + for (int i = 0; i < maxCycles; i++) { - cluster.get(1).runOnInstance(() -> { + cluster.get(1).acceptOnInstance((ks, tbl) -> { + Keyspace.open(ks).getColumnFamilyStore(tbl).forceBlockingFlush(UNIT_TESTS); + Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.COMMANDS_FOR_KEY).forceBlockingFlush(UNIT_TESTS); Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.JOURNAL).forceMajorCompaction(); - }); + }, schema.keyspace, schema.table); after = countDiffs.call(); if (after == 0) return; - Thread.sleep(1000); + Thread.sleep(10000); } - Assert.fail("Should have GC'd all in (way under) 60 cycles. Remaining: " + after); + Assert.fail("Should have GC'd all in (way under) " + maxCycles + " cycles. Remaining: " + after); }); } }