Serialise local metadata proposals so concurrent schema changes do not exhaust their retry budget contending for the same epoch

Concurrent CQL DDL submitted to a CMS member could fail with a SERVER_ERROR while
the request deadline still had seconds remaining. Every concurrent commit read the
same highest consecutive epoch and proposed at epoch+1, so all but one loser per
round burned a retry attempt, backed off and refetched the log before recomputing.
With the attempt limit from the default cms_retry_delay that budget was exhausted
long before request_timeout, so valid statements were rejected while several
seconds of their deadline remained.

Since tryCommitOne can only ever apply one proposal per epoch, and Paxos already
takes a per-partition coordinator lock for the duration of the CAS, racing that
slot from many local request threads cannot increase throughput; it only converts
would-be waiting into wasted attempts. AbstractLocalProcessor now serialises the
read-execute-propose cycle for locally originating commits behind a fair lock
whose acquisition is bounded by the caller's own deadline. The lock is released
before the follower wait, the backoff sleep and any log fetch, so none of those
block another proposer. A request which cannot be admitted in time reports that
distinctly, and only claims that nothing was proposed when no earlier iteration
had reached the CAS, since a lost epoch and an ambiguous Paxos result are
indistinguishable to the caller.

This does not address races between separate CMS members, which remain the
distributed log's responsibility. Admission is FIFO and does not prioritise by
transformation kind, so a topology change can queue behind schema DDL which
arrived first.

patch by Patrick McFadin; reviewed by TBD for CASSANDRA-21537

Assisted-by: Claude Opus 5 <noreply@anthropic.com>

Author: pmcfadin

CHANGES.txt

@@ -1,4 +1,5 @@
 7.0
+ * Serialise local metadata proposals so concurrent schema changes do not exhaust their retry budget contending for the same epoch (CASSANDRA-21537)
  * Don't increment client metrics on messaging service connection unpause (CASSANDRA-21491)
  * Add nodetool getreplicas (CASSANDRA-17665)
  * Implementation of CEP-49: Hardware-accelerated compression (CASSANDRA-20975)

src/java/org/apache/cassandra/tcm/AbstractLocalProcessor.java

@@ -18,6 +18,7 @@
 
 package org.apache.cassandra.tcm;
 
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Supplier;
 
 import org.slf4j.Logger;
@@ -41,6 +42,44 @@ public abstract class AbstractLocalProcessor implements Processor
 
     protected final LocalLog log;
 
+    /**
+     * Serialises the read-execute-propose cycle for commits originating on this node, i.e.
+     * {@link LocalLog#waitForHighestConsecutive}, execution of the transformation, the {@link #tryCommitOne} proposal
+     * and, when that proposal wins its epoch, the subsequent {@link LocalLog#append}.
+     *
+     * Invariants:
+     * <ul>
+     *     <li>acquisition waits at most the caller's remaining {@link Retry} deadline, so being queued consumes only
+     *         the waiter's own budget and can never push it past its own deadline;</li>
+     *     <li>every successful acquisition is matched by exactly one unlock;</li>
+     *     <li>it is held across one whole cycle, including the blocking and remote work that cycle needs, and is
+     *         released before everything which is not part of deriving and proposing an epoch: local enactment via
+     *         {@link LocalLog#awaitAtLeast}, the uninterruptible retry backoff, and any log fetch performed after a
+     *         rejection or a lost epoch;</li>
+     *     <li>admission is fair, hence FIFO, and does not prioritise by {@link Transformation.Kind}: a topology
+     *         change can queue behind schema DDL which arrived first.</li>
+     * </ul>
+     *
+     * This does not, and cannot, address races between separate CMS members; those remain the distributed log's job.
+     */
+    private final ReentrantLock proposalLock = new ReentrantLock(true);
+
+    /**
+     * What a single read-execute-propose cycle concluded, so that the cycle can hand off to the code which acts on it
+     * after {@link #proposalLock} has been released.
+     */
+    private enum Outcome
+    {
+        /** The proposal won its epoch and has been appended to the local log. */
+        APPLIED,
+        /** Execution against the latest local metadata was rejected; nothing was proposed. */
+        REJECTED,
+        /** The proposal did not win its epoch, or its outcome is unknown. */
+        NOT_APPLIED,
+        /** The proposal threw; the throwable is reported and inspected once the lock has been released. */
+        FAILED
+    }
+
     public AbstractLocalProcessor(LocalLog log)
     {
         this.log = log;
@@ -57,35 +96,99 @@ public final Commit.Result commit(Entry.Id entryId, Transformation transform, fi
         String transformStr = transform.toString(); // convert once as idempotent and used in multiple logs
         logger.debug("Starting local commit of {} with policy {}", transformStr, retryPolicy);
         long commitStart = nanoTime();
+        // History, not per-iteration state: set immediately before the first tryCommitOne call and never cleared, so
+        // that the failure message below can only claim nothing was proposed while it is still false.
+        boolean proposalAttempted = false;
         while (!retryPolicy.hasExpired())
         {
-            ClusterMetadata previous = log.waitForHighestConsecutive();
-            if (!acceptCommit(previous))
-            {
-                String msg = String.format("Node %s is not a CMS member in epoch %s; members=%s",
-                                           FBUtilities.getBroadcastAddressAndPort(),
-                                           previous.epoch.getEpoch(),
-                                           previous.fullCMSMembers());
-                logger.warn(msg);
-                throw new NotCMSException(msg);
-            }
+            // Serialise the read-execute-propose cycle for locally originating commits. Acquisition consumes this
+            // caller's own deadline, so a request which cannot be admitted in time gives up here rather than proposing.
+            if (!acquireProposalLock(retryPolicy))
+                return notAdmitted(transformStr, commitStart, retryPolicy, proposalAttempted);
 
+            ClusterMetadata previous;
             Transformation.Result result;
-            if (!transform.eligibleToCommit(previous))
+            Outcome outcome;
+            Throwable failure = null;
+            // The one critical section: read the highest consecutive metadata, execute against it, propose the derived
+            // epoch and, if the proposal won, append it locally. Nothing else belongs here, and the lock is dropped in
+            // the single finally below before any of the outcomes are acted on.
+            try
             {
-                result = new Transformation.Rejected(INVALID, "Transformation rejected, can't commit " + transformStr +
-                                                              " it not supported with cluster common serialization version " + previous.directory.commonSerializationVersion +
-                                                              " and min/max serialization versions " + previous.directory.clusterMinVersion + "/" + previous.directory.clusterMaxVersion);
+                previous = log.waitForHighestConsecutive();
+                if (!acceptCommit(previous))
+                {
+                    String msg = String.format("Node %s is not a CMS member in epoch %s; members=%s",
+                                               FBUtilities.getBroadcastAddressAndPort(),
+                                               previous.epoch.getEpoch(),
+                                               previous.fullCMSMembers());
+                    logger.warn(msg);
+                    throw new NotCMSException(msg);
+                }
+
+                if (!transform.eligibleToCommit(previous))
+                {
+                    result = new Transformation.Rejected(INVALID, "Transformation rejected, can't commit " + transformStr +
+                                                                  " it not supported with cluster common serialization version " + previous.directory.commonSerializationVersion +
+                                                                  " and min/max serialization versions " + previous.directory.clusterMinVersion + "/" + previous.directory.clusterMaxVersion);
+                }
+                else
+                {
+                    result = executeStrictly(previous, transform);
+                }
+
+                if (result.isRejected())
+                {
+                    // Nothing to propose; the catch-up which decides whether this rejection is final happens after the
+                    // lock is released, as it may involve remote peers.
+                    outcome = Outcome.REJECTED;
+                }
+                else
+                {
+                    try
+                    {
+                        Epoch nextEpoch = result.success().metadata.epoch;
+                        // If metadata applies, try committing it to the log
+                        long casStart = nanoTime();
+                        proposalAttempted = true;
+                        boolean applied = tryCommitOne(entryId, transform, previous.epoch, nextEpoch);
+                        long casElapsedUs = NANOSECONDS.toMicros(nanoTime() - casStart);
+                        logger.debug("tryCommitOne for {} epoch {}->{}: applied={}, took {}us",
+                                     transform.kind(), previous.epoch, nextEpoch, applied, casElapsedUs);
+
+                        // Application here semantially means "succeeded in committing to the distributed log".
+                        if (applied)
+                        {
+                            logger.info("Committed {}. New epoch is {}. Took {} attempts in {}us total.",
+                                        transformStr, nextEpoch, retryPolicy.attempts(),
+                                        NANOSECONDS.toMicros(nanoTime() - commitStart));
+                            log.append(new Entry(entryId, nextEpoch, new Transformation.Executed(transform, result)));
+                            outcome = Outcome.APPLIED;
+                        }
+                        else
+                        {
+                            outcome = Outcome.NOT_APPLIED;
+                        }
+                    }
+                    catch (Throwable e)
+                    {
+                        // Reported and inspected outside the lock, so that neither is done while holding it.
+                        failure = e;
+                        outcome = Outcome.FAILED;
+                    }
+                }
             }
-            else
+            finally
             {
-                result = executeStrictly(previous, transform);
+                proposalLock.unlock();
             }
 
-            // If we got a rejection, it could be that _we_ are not aware of the highest epoch.
-            // Just try to catch up to the latest distributed state.
-            if (result.isRejected())
+            // Everything from here on runs without the lock: it is either remote, uninterruptible, or waits on the
+            // local log being caught up, and none of it is part of deriving and proposing an epoch.
+            if (outcome == Outcome.REJECTED)
             {
+                // If we got a rejection, it could be that _we_ are not aware of the highest epoch.
+                // Just try to catch up to the latest distributed state.
                 // Use a dedicated retry policy here as the one for the commit itself may not be appropriate.
                 // It uses the wrong metric and for STARTUP transformations will retry indefinitely, which is not
                 // what we want here.
@@ -107,43 +210,43 @@ public final Commit.Result commit(Entry.Id entryId, Transformation transform, fi
                 continue;
             }
 
+            if (outcome == Outcome.FAILED)
+            {
+                logger.error("Caught error while trying to perform a local commit", failure);
+                JVMStabilityInspector.inspectThrowable(failure);
+                if (!retryPolicy.maybeSleep())
+                    break;
+                continue;
+            }
+
             try
             {
-                Epoch nextEpoch = result.success().metadata.epoch;
-                // If metadata applies, try committing it to the log
-                long casStart = nanoTime();
-                boolean applied = tryCommitOne(entryId, transform, previous.epoch, nextEpoch);
-                long casElapsedUs = NANOSECONDS.toMicros(nanoTime() - casStart);
-                logger.debug("tryCommitOne for {} epoch {}->{}: applied={}, took {}us",
-                             transform.kind(), previous.epoch, nextEpoch, applied, casElapsedUs);
-
-                // Application here semantially means "succeeded in committing to the distributed log".
-                if (applied)
+                if (outcome == Outcome.APPLIED)
                 {
-                    logger.info("Committed {}. New epoch is {}. Took {} attempts in {}us total.",
-                                transformStr, nextEpoch, retryPolicy.attempts(),
-                                NANOSECONDS.toMicros(nanoTime() - commitStart));
-                    log.append(new Entry(entryId, nextEpoch, new Transformation.Executed(transform, result)));
+                    // The proposal is durable in the distributed log and appended locally, so the next proposer can
+                    // already derive its epoch from it. Holding the lock across this wait would serialise the local
+                    // enactment latency of this commit onto unrelated ones.
+                    Epoch nextEpoch = result.success().metadata.epoch;
                     log.awaitAtLeast(nextEpoch);
 
-                    return new Commit.Result.Success(result.success().metadata.epoch,
+                    return new Commit.Result.Success(nextEpoch,
                                                      toLogState(result.success(), entryId, lastKnown, transform));
                 }
-                else
-                {
-                    if (!retryPolicy.maybeSleep())
-                        break;
-
-                    logger.info("Backed off after failure to commit to log, fetching latest log entries before retry");
-                    // TODO: could also add epoch from mis-application from [applied].
-                    // Use a dedicated retry policy here as the one for the commit itself may not be appropriate.
-                    // It uses the wrong metric and for STARTUP transformations will retry indefinitely, which is not
-                    // what we want here.
-                    Retry fetchLogRetry = Retry.until(retryPolicy.deadlineNanos, TCMMetrics.instance.fetchLogRetries);
-                    fetchLogAndWait(null, fetchLogRetry);
-                    logger.info("Fetched latest log entries, re-entering commit retry loop with {}ms remaining until deadline",
-                                NANOSECONDS.toMillis(retryPolicy.remainingNanos()));
-                }
+
+                // Lost the epoch, or the outcome is unknown. Back off, then refetch: the sleep is uninterruptible and
+                // the fetch may be remote, neither of which should block another local proposer.
+                if (!retryPolicy.maybeSleep())
+                    break;
+
+                logger.info("Backed off after failure to commit to log, fetching latest log entries before retry");
+                // TODO: could also add epoch from mis-application from [applied].
+                // Use a dedicated retry policy here as the one for the commit itself may not be appropriate.
+                // It uses the wrong metric and for STARTUP transformations will retry indefinitely, which is not
+                // what we want here.
+                Retry fetchLogRetry = Retry.until(retryPolicy.deadlineNanos, TCMMetrics.instance.fetchLogRetries);
+                fetchLogAndWait(null, fetchLogRetry);
+                logger.info("Fetched latest log entries, re-entering commit retry loop with {}ms remaining until deadline",
+                            NANOSECONDS.toMillis(retryPolicy.remainingNanos()));
             }
             catch (Throwable e)
             {
@@ -153,14 +256,73 @@ public final Commit.Result commit(Entry.Id entryId, Transformation transform, fi
                     break;
             }
         }
-        long remainingMillis = NANOSECONDS.toMillis(retryPolicy.remainingNanos());
         String failureMsg = String.format("Could not perform commit after %d attempts. Time remaining: %dms",
-                                          retryPolicy.attempts(), remainingMillis);
+                                          retryPolicy.attempts(), NANOSECONDS.toMillis(retryPolicy.remainingNanos()));
+        return failed(transformStr, commitStart, failureMsg);
+    }
+
+    /**
+     * Builds the failure for a commit which could not be admitted to {@link #proposalLock} within its deadline. Either
+     * shutdown or a cancelled request, or simply too many local commits queued ahead of this one.
+     *
+     * @param proposalAttempted whether any earlier iteration already reached {@link #tryCommitOne}. If it did, its
+     *        outcome may be genuinely unknown, as losing the epoch and an ambiguous Paxos result are indistinguishable
+     *        to us, so the message states only what happened and promises nothing about whether the transformation
+     *        applied. If it did not, retrying is unambiguously safe, and reporting an attempt count would imply CAS
+     *        activity which did not happen.
+     */
+    private Commit.Result notAdmitted(String transformStr, long commitStart, Retry retryPolicy, boolean proposalAttempted)
+    {
+        String cause = Thread.currentThread().isInterrupted()
+                       ? "Interrupted while waiting to submit commit"
+                       : String.format("Timed out after %dms waiting to submit commit",
+                                       NANOSECONDS.toMillis(nanoTime() - commitStart));
+        long remainingMillis = NANOSECONDS.toMillis(retryPolicy.remainingNanos());
+        String failureMsg = proposalAttempted
+                            ? String.format("%s. A proposal was made during one of the preceding %d attempts, so " +
+                                            "whether the transformation was committed to the log is unknown. " +
+                                            "Time remaining: %dms",
+                                            cause, retryPolicy.attempts(), remainingMillis)
+                            : String.format("%s. No proposal was made. Time remaining: %dms", cause, remainingMillis);
+        return failed(transformStr, commitStart, failureMsg);
+    }
+
+    private Commit.Result failed(String transformStr, long commitStart, String failureMsg)
+    {
         logger.debug("Commit {} failed in {}us total. {}",
                      transformStr, NANOSECONDS.toMicros(nanoTime() - commitStart), failureMsg);
         return Commit.Result.failed(SERVER_ERROR, failureMsg);
     }
 
+    /**
+     * Acquires {@link #proposalLock}, waiting no longer than the caller's remaining deadline. Every successful
+     * acquisition is matched by exactly one unlock in the {@code finally} which closes the critical section, so
+     * hold-count accounting is safe on a thread which is already mid-cycle: a nested acquisition only raises the
+     * count, and releasing it cannot free the lock out from under the outer cycle. That is an accounting invariant
+     * only; it does not assert that a nested commit is semantically safe, as a synchronous pre-commit listener could
+     * recursively encounter the same pending entry. No current production listener does this.
+     *
+     * @return true if the lock is held by the calling thread on return, false if the deadline expired first or the
+     *         thread was interrupted while waiting, in which case this attempt proposed nothing.
+     */
+    private boolean acquireProposalLock(Retry retryPolicy)
+    {
+        long remainingNanos = retryPolicy.remainingNanos();
+        if (remainingNanos <= 0)
+            return false;
+
+        try
+        {
+            return proposalLock.tryLock(remainingNanos, NANOSECONDS);
+        }
+        catch (InterruptedException e)
+        {
+            // Preserve the interrupt for the caller (e.g. so shutdown is not swallowed) and give up on this attempt.
+            Thread.currentThread().interrupt();
+            return false;
+        }
+    }
+
     public Commit.Result maybeFailure(Entry.Id entryId, Epoch lastKnown, Supplier<Commit.Result.Failure> orElse)
     {
         LogState logState = toLogState(lastKnown);