Rerebootstrap

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
-	url = https://github.com/apache/cassandra-accord.git
-	branch = trunk
+	url = https://github.com/belliottsmith/cassandra-accord.git
+	branch = rerebootstrap

UNFINISHED.md

@@ -0,0 +1,28 @@
+### D1. Partial log unavailability (the piece you're deferring)
+
+ Current state, so it's recoverable later:
+ - Cleanup.isUnavailable (Cleanup.java:229-233) — any(LOG_UNAVAILABLE) or all(LOG_INCOMPLETE) && saveStatus < Stable, with your TODO (required): how does this interact with lost ownership?.
+ - SafeCommandStore.registerTransitiveDeps (~:565) — per-range subtraction via redundantBefore.removeLogUnavailableOrIncomplete(txnId, tmp), with TODO (required): if we only part-filter we're still going to have problems, as we won't be able to
+   update the transaction.
+ - RedundantBefore.Bounds.depBound — folds LOG_INCOMPLETE/LOG_UNAVAILABLE into the dep bound when UNREADY is ahead of the applied bound; this is what keeps deps sound after a restart, when the in-memory refuses map is gone.
+ - MapReduceCommandStores.Refuse{NONE,DEPS,ALL} + per-message abort() — the in-memory, coarse (whole-scope) refusal, only alive between unsafeRefuseRequests and unsafeAcceptRequests.
+
+ These four must agree on one predicate. Concretely the questions to settle:
+ 1. Is any ever right for LOG_UNAVAILABLE? The dangerous arm isn't FULL (throwing is merely unavailability) but non-FULL, where logUnavailable(...) returns ERASE — a compaction decision that discards the record for ranges whose log is intact.
+    If nothing else changes, splitting just that arm (throw on any, erase only on all) is strictly safer and is a one-line intermediate.
+ 2. Lost ownership / retired ranges break all. Ranges that are WAS_OWNED or locally retired never carry LOG_INCOMPLETE, so all(...) is false and we don't refuse even though the owned remainder is incomplete. Your "treat pre-log-point ranges as
+    locally-retired-like" idea is exactly the missing normalisation, and note the patch already does the dual of it in SafeCommandStore.refuses(participants) (removeLocallyRetired(participants) before folding). A shared helper — "the
+    participants we must have a complete log for" = participants − retired − wasOwned − logIncomplete/Unavailable — would let Cleanup, refuses() and registerTransitive share one definition instead of three.
+ 3. Recovery cannot answer partially. BeginRecovery refuses on refuses.max != NONE, which only covers the in-memory window, not the post-restart window. There is no wire representation for "valid only for a sub-range" — but ReadOk.unavailable
+    already is exactly that for reads, so a RecoverNack{unavailable} (and the analogous field on the deps replies) is the smallest precedent-following addition; it would also let Commit/Apply stop depending on the coarse Refuse.MinMax.
+ 4. Future distinction you mentioned: if you do split further, the axis that matters to consumers looks like "records absent below the bound" (incomplete) vs "records possibly wrong below the bound" (corrupted) vs "records absent but decisions
+    recoverable from peers". Worth writing that down in BootstrapReason/RedundantStatus.Property javadoc before adding a third status, since the persisted bit space is nearly full (see N9).
+
+ ### D2. Empty / short DurabilityResults
+
+ Now benign-ish but still wrong: DurabilityService.submit's canUseDurableBefore && isAlreadyMet shortcut calls reportSuccess() while success == DurabilityResults.EMPTY, so markBootstrapping marks nothing and the attempt completes with missing
+ == valid → onFailedBootstrap → retry. Since each retry re-derives a fresh maxConflict and can be satisfied by durableBefore again, an idle range can retry indefinitely, with the unhelpful "Unknown failure" from complete()'s fail runnable.
+ Options: (a) populate DurabilityResults.of(ranges, request.min, <nodes>) on the shortcut path; (b) make the shortcut ineligible when the requester needs per-range bounds/readable (a flag on the request); (c) fail prepareToBootstrap explicitly
+ if byTxnId() doesn't cover the requested ranges, so the diagnostic names the cause. Also sync(...)/close(...) still return success(null) for empty ranges — an empty DurabilityResults would be safer for every caller that dereferences the value.
+ Worth adding an Invariants.require(success.byTxnId() covers commitRanges) at the markBootstrapping site so this class of shortfall is loud rather than a retry loop.
+

doc/modules/cassandra/pages/architecture/accord-architecture.adoc

@@ -87,7 +87,7 @@ finally {
 }
 ....
 
-In other words, `CommandStore` collects the `PreLoadContext`, state
+In other words, `CommandStore` collects the `ExecutionContext`, state
 required to be in memory for command execution (possible dependencies,
 such as `TxnId`s, and `Key`s of commands, but also `CommandsForKeys`
 that will be needed during execution). Once the context is collected and
@@ -103,11 +103,11 @@ to ensure transactional integrity, changes to commands are tracked and
 are recorded into `Journal` for crash-recovery. `ProgressLog` and
 `CommandsForKey` are up
 
-On Cassandra side, concurrent execution is controlled by `AccordTask`,
+On Cassandra side, concurrent execution is controlled by `SafeTask`,
 which contains cache loading logic and persistence callbacks. Since
 Accord may potentially hold a large number of command states in memory,
 their states may be _shrunk_ to their binary representation to save some
-memory, or they can get fully evicted. This also means that `AccordTask`
+memory, or they can get fully evicted. This also means that `SafeTask`
 will have to reload relevant dependencies from preload context before
 command execution can begin.
 

src/java/org/apache/cassandra/concurrent/CassandraThread.java

@@ -18,14 +18,24 @@
 
 package org.apache.cassandra.concurrent;
 
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
 import org.apache.cassandra.metrics.ThreadLocalMetrics;
+import org.apache.cassandra.service.accord.execution.AccordExecutor;
+import org.apache.cassandra.service.accord.execution.Task;
+import org.apache.cassandra.service.accord.execution.TaskRunner;
 
 import io.netty.util.concurrent.FastThreadLocalThread;
 
-public class CassandraThread extends FastThreadLocalThread
+public class CassandraThread extends FastThreadLocalThread implements TaskRunner
 {
     private ThreadLocalMetrics threadLocalMetrics;
     private ExecutorLocals executorLocals;
+    private AccordExecutor accordActiveExecutor;
+    private AccordExecutor accordLockedExecutor;
+    private int accordLockedExecutorDepth;
+    private volatile Task accordActiveTask;
+    private static final AtomicReferenceFieldUpdater<CassandraThread, Task> accordActiveTaskUpdater = AtomicReferenceFieldUpdater.newUpdater(CassandraThread.class, Task.class, "accordActiveTask");
 
     private final ImmediateTaskHolder immediateTaskHolder;
 
@@ -89,6 +99,55 @@ public ExecutorLocals replaceExecutorLocals(ExecutorLocals newExecutorLocals)
         return current != null ? current : ExecutorLocals.none();
     }
 
+    public final AccordExecutor accordActiveExecutor()
+    {
+        return accordActiveExecutor;
+    }
+
+    public final void setAccordActiveExecutor(AccordExecutor newExecutor)
+    {
+        accordActiveExecutor = newExecutor;
+    }
+
+    @Override
+    public final AccordExecutor accordLockedExecutor()
+    {
+        return accordLockedExecutor;
+    }
+
+    @Override
+    public final boolean tryEnterAccordLockedExecutor(AccordExecutor newLockedExecutor)
+    {
+        if (accordLockedExecutor == null) accordLockedExecutor = newLockedExecutor;
+        else if (accordLockedExecutor != newLockedExecutor) return false;
+        ++accordLockedExecutorDepth;
+        return true;
+    }
+
+    @Override
+    public final void exitAccordLockedExecutor()
+    {
+        if (--accordLockedExecutorDepth == 0)
+            accordLockedExecutor = null;
+    }
+
+    public final Task accordActiveTask()
+    {
+        return accordActiveTask;
+    }
+
+    // to be called only by the thread itself, so can (eventually) avoid any memory barriers
+    public final Task accordActiveSelfTask()
+    {
+        // TODO (expected): with newer JDK use accordActiveTaskUpdater.getPlain
+        return accordActiveTask;
+    }
+
+    public final void setAccordActiveTask(Task newActiveTask)
+    {
+        accordActiveTaskUpdater.lazySet(this, newActiveTask);
+    }
+
     // final to avoid skipping of the cleanup logic in child classes
     final public void run()
     {

src/java/org/apache/cassandra/concurrent/Stage.java

@@ -47,7 +47,7 @@ public enum Stage
     MUTATION           (true,  "MutationStage",         "request",  DatabaseDescriptor::getConcurrentWriters,        DatabaseDescriptor::setConcurrentWriters,        Stage::multiThreadedLowSignalStage),
     COUNTER_MUTATION   (true,  "CounterMutationStage",  "request",  DatabaseDescriptor::getConcurrentCounterWriters, DatabaseDescriptor::setConcurrentCounterWriters, Stage::multiThreadedLowSignalStage),
     VIEW_MUTATION      (true,  "ViewMutationStage",     "request",  DatabaseDescriptor::getConcurrentViewWriters,    DatabaseDescriptor::setConcurrentViewWriters,    Stage::multiThreadedLowSignalStage),
-    ACCORD_MIGRATION   (false, "AccordMigrationStage", "request",   DatabaseDescriptor::getAccordConcurrentOps,      DatabaseDescriptor::setConcurrentAccordOps,      Stage::multiThreadedLowSignalStage),
+    ACCORD_MIGRATION   (false, "AccordMigrationStage",  "request",  DatabaseDescriptor::getAccordConcurrentMigrationOps, DatabaseDescriptor::setConcurrentAccordMigrationOps,             Stage::multiThreadedLowSignalStage),
     GOSSIP             (true,  "GossipStage",           "internal", () -> 1,                                         null,                                            Stage::singleThreadedStage),
     REQUEST_RESPONSE   (false, "RequestResponseStage",  "request",  FBUtilities::getAvailableProcessors,             null,                                            Stage::multiThreadedLowSignalStage),
     ANTI_ENTROPY       (false, "AntiEntropyStage",      "internal", () -> 1,                                         null,                                            Stage::singleThreadedStage),