CEP-45: BRR does tracked writes during migration to mutation tracking

During migration from untracked to tracked replication, blocking read
repair (BRR) continues to run because mutation tracking lacks sufficient
initialized state to provide monotonic reads for pending ranges. However,
BRR was previously bypassing mutation tracking entirely — read repair
mutations were sent as untracked writes via READ_REPAIR_REQ, which meant
they were invisible to the mutation journal. This is unsafe because the
migration repair that completes the transition assumes all writes after it
started are tracked, so that tracked reads can safely begin afterward.

Route BRR writes through TrackedWriteRequest.perform() when
MigrationRouter.shouldUseTrackedForWrites() indicates the token is
migrating. This gives each read repair mutation a proper MutationId and
records it in the mutation journal, ensuring the migration invariant holds.

Key changes:
- Remove the isReadRepair flag from Mutation and all special-case bypasses
  that let read repair skip mutation tracking validation and routing
- Add repairViaTrackedWrite() to BlockingReadRepair which sends each
  per-replica mutation as an independent tracked write with retry logic
  for migration races (RetryOnDifferentSystemException/CoordinatorBehindException)
- Introduce WriteCallback interface (replacing Runnable) on write response
  handlers so tracked write completion can be observed without races
- Make tracked→untracked instant in AlterSchema by skipping migration
  state entirely — tracked writes are already quorum writes so untracked
  reads are strictly less demanding. Reverting an in-progress
  untracked→tracked migration cleans up the now-unnecessary state.
- Add RepairedBlockingViaTrackedWrite metric to observe BRR-via-tracked path
- Add pauseRegularPriorityForTesting() to ActiveLogReconciler to allow
  tests to suppress background reconciliation while keeping high-priority
  (tracked read) reconciliation active
- Add MutationTrackingReadRepairTest covering BRR behavior across all
  migration phases with both point reads and range scans

Author: aweisberg

src/java/org/apache/cassandra/db/AbstractMutationVerbHandler.java

@@ -202,13 +202,6 @@ else if (message.epoch().isBefore(metadata.schema.lastModified()))
      */
     protected ClusterMetadata checkReplicationMigration(ClusterMetadata metadata, Message<T> message, InetAddressAndPort respondTo)
     {
-        // Read repair mutations always bypass mutation tracking and use the untracked
-        // write path, so skip the replication migration routing check. The isReadRepair
-        // flag on the mutation hasn't been set yet at this point — it's set later in
-        // applyMutation() — so we check the handler type instead.
-        if (this instanceof ReadRepairVerbHandler)
-            return metadata;
-
         IMutation mutation = message.payload;
         MutationRouting expected = mutation.id().isNone() ? MutationRouting.UNTRACKED : MutationRouting.TRACKED;
         if (expected == MigrationRouter.getMutationRouting(metadata, mutation))

src/java/org/apache/cassandra/db/CassandraKeyspaceWriteHandler.java

@@ -47,8 +47,7 @@ public WriteContext beginWrite(Mutation mutation, boolean makeDurable) throws Re
         {
             group = Keyspace.writeOrder.start();
 
-            if (!mutation.isReadRepair())
-                MigrationRouter.validateUntrackedMutation(mutation);
+            MigrationRouter.validateUntrackedMutation(mutation);
             // write the mutation to the commitlog and memtables
             CommitLogPosition position = null;
             if (makeDurable)

src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java

@@ -70,7 +70,7 @@ protected void applyMutation(final Message<CounterMutation> message, InetAddress
         // it's own in that case.
         StorageProxy.applyCounterMutationOnLeader(cm,
                                                   localDataCenter,
-                                                  () -> MessagingService.instance().send(message.emptyResponse(), respondToAddress),
+                                                  handler -> MessagingService.instance().send(message.emptyResponse(), respondToAddress),
                                                   Dispatcher.RequestTime.forImmediateExecution());
     }
 }

src/java/org/apache/cassandra/db/Keyspace.java

@@ -439,7 +439,7 @@ public void apply(final Mutation mutation,
                       boolean updateIndexes,
                       boolean isDroppable)
     {
-        if (MigrationRouter.isFullyTracked(mutation) && !mutation.isReadRepair())
+        if (MigrationRouter.isFullyTracked(mutation))
             applyInternalTracked(mutation, null);
         else
             applyInternal(mutation, makeDurable, updateIndexes, isDroppable, false, null);
@@ -462,7 +462,7 @@ private Future<?> applyInternal(final Mutation mutation,
                                                boolean isDeferrable,
                                                Promise<?> future)
     {
-        Preconditions.checkState((!getMetadata().useMutationTracking() || mutation.isReadRepair()) && mutation.id().isNone());
+        Preconditions.checkState(!MigrationRouter.isFullyTracked(mutation) && mutation.id().isNone());
 
         if (TEST_FAIL_WRITES && getMetadata().name.equals(TEST_FAIL_WRITES_KS))
             throw new RuntimeException("Testing write failures");

src/java/org/apache/cassandra/db/Mutation.java

@@ -110,21 +110,6 @@ public class Mutation implements IMutation, Supplier<Mutation>, Commitable
     // because it is being applied by one or in a context where transaction conflicts don't occur
     private PotentialTxnConflicts potentialTxnConflicts;
 
-    // Transient: not serialized on the wire. Set by ReadRepairVerbHandler on the
-    // receiving side so downstream code (Keyspace.apply, write handlers) can route
-    // read repair mutations through the untracked write path during migration.
-    private transient boolean isReadRepair;
-
-    public void setReadRepair(boolean readRepair)
-    {
-        this.isReadRepair = readRepair;
-    }
-
-    public boolean isReadRepair()
-    {
-        return isReadRepair;
-    }
-
     public Mutation(MutationId id, PartitionUpdate update)
     {
         this(id, update.metadata().keyspace, update.partitionKey(), ImmutableMap.of(update.metadata().id, update), approxTime.now(), update.metadata().params.cdc, PotentialTxnConflicts.DISALLOW);
@@ -165,9 +150,7 @@ public MutationId id()
     @Override
     public Mutation withMutationId(MutationId mutationId)
     {
-        Mutation m = new Mutation(mutationId, keyspaceName, key, modifications, approxCreatedAtNanos, cdcEnabled, potentialTxnConflicts);
-        m.isReadRepair = this.isReadRepair;
-        return m;
+        return new Mutation(mutationId, keyspaceName, key, modifications, approxCreatedAtNanos, cdcEnabled, potentialTxnConflicts);
     }
 
     private static boolean cdcEnabled(Iterable<PartitionUpdate> modifications)
@@ -201,9 +184,7 @@ private static boolean cdcEnabled(Iterable<PartitionUpdate> modifications)
 
         Map<TableId, PartitionUpdate> updates = builder.build();
         checkState(!updates.isEmpty(), "Updates should not be empty");
-        Mutation result = new Mutation(id, keyspaceName, key, builder.build(), approxCreatedAtNanos, potentialTxnConflicts);
-        result.isReadRepair = this.isReadRepair;
-        return result;
+        return new Mutation(id, keyspaceName, key, builder.build(), approxCreatedAtNanos, potentialTxnConflicts);
     }
 
     public @Nullable Mutation without(TableId tableId)

src/java/org/apache/cassandra/db/ReadRepairVerbHandler.java

@@ -27,7 +27,6 @@ public class ReadRepairVerbHandler extends AbstractMutationVerbHandler<Mutation>
 
     public void applyMutation(Mutation mutation)
     {
-        mutation.setReadRepair(true);
         mutation.apply();
     }
 

src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 import com.google.common.base.Preconditions;
@@ -93,7 +94,7 @@ protected AbstractReplicationStrategy(String keyspaceName, Map<String, String> c
     public abstract DataPlacement calculateDataPlacement(Epoch epoch, List<Range<Token>> ranges, ClusterMetadata metadata);
 
     public <T> AbstractWriteResponseHandler<T> getWriteResponseHandler(ReplicaPlan.ForWrite replicaPlan,
-                                                                       Runnable callback,
+                                                                       Consumer<AbstractWriteResponseHandler<?>> callback,
                                                                        WriteType writeType,
                                                                        Supplier<Mutation> hintOnFailure,
                                                                        Dispatcher.RequestTime requestTime)
@@ -103,7 +104,7 @@ public <T> AbstractWriteResponseHandler<T> getWriteResponseHandler(ReplicaPlan.F
     }
 
     public <T> AbstractWriteResponseHandler<T> getWriteResponseHandler(ReplicaPlan.ForWrite replicaPlan,
-                                                                       Runnable callback,
+                                                                       Consumer<AbstractWriteResponseHandler<?>> callback,
                                                                        WriteType writeType,
                                                                        Supplier<Mutation> hintOnFailure,
                                                                        Dispatcher.RequestTime requestTime,

src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java

@@ -37,6 +37,12 @@ public class ReadRepairMetrics
      */
     public static final Meter repairedBlockingViaAccord = Metrics.meter(factory.createMetricName("RepairedBlockingViaAccord"));
 
+    /**
+     * Blocking read repair was sent as a tracked write. This happens when the token is migrating to tracked replication
+     * but migration hasn't completed
+     */
+    public static final Meter repairedBlockingViaTrackedWrite = Metrics.meter(factory.createMetricName("RepairedBlockingViaTrackedWrite"));
+
     /**
      * This should be zero if you are trying to run Accord in a 100% correct way and interoperating with non-transactional writes.
      *

src/java/org/apache/cassandra/replication/ActiveLogReconciler.java

@@ -120,8 +120,11 @@ public void run(Interruptible.State state) throws InterruptedException
             Task task;
             while ((task = highPriorityTasks.poll()) != null)
                 task.send();
-            while ((task = regularPriorityTasks.poll()) != null)
-                task.send();
+            if (!isRegularPriorityPaused)
+            {
+                while ((task = regularPriorityTasks.poll()) != null)
+                    task.send();
+            }
 
             haveWork.acquire(1);
         }
@@ -261,6 +264,7 @@ void send()
 
     private volatile boolean isShutdown = false;
     private volatile boolean isPaused = false;
+    private volatile boolean isRegularPriorityPaused = false;
 
     @Override
     public boolean isTerminated()
@@ -307,4 +311,22 @@ void resumeForTesting()
     {
         isPaused = false;
     }
+
+    /**
+     * Pause only regular-priority (background write retry) task delivery.
+     * High-priority tasks (needed by tracked read reconciliation) continue to be processed.
+     * This allows tests to prevent the reconciler from proactively fixing inconsistencies
+     * while still allowing tracked reads to pull missing mutations.
+     */
+    @VisibleForTesting
+    void pauseRegularPriorityForTesting()
+    {
+        isRegularPriorityPaused = true;
+    }
+
+    @VisibleForTesting
+    void resumeRegularPriorityForTesting()
+    {
+        isRegularPriorityPaused = false;
+    }
 }

src/java/org/apache/cassandra/replication/ForwardedWrite.java

@@ -24,6 +24,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Consumer;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Sets;
@@ -297,6 +298,13 @@ private static void applyLocallyAndForwardToReplicas(Mutation mutation, Set<Node
     }
 
     public static AbstractWriteResponseHandler<Object> forwardMutation(Mutation mutation, ReplicaPlan.ForWrite plan, AbstractReplicationStrategy strategy, Dispatcher.RequestTime requestTime)
+    {
+        return forwardMutationInternal(mutation, plan, strategy, requestTime, null);
+    }
+
+    private static AbstractWriteResponseHandler<Object> forwardMutationInternal(
+        Mutation mutation, ReplicaPlan.ForWrite plan, AbstractReplicationStrategy strategy,
+        Dispatcher.RequestTime requestTime, Consumer<AbstractWriteResponseHandler<?>> onComplete)
     {
         // find leader
         NodeProximity proximity = DatabaseDescriptor.getNodeProximity();
@@ -318,7 +326,7 @@ public static AbstractWriteResponseHandler<Object> forwardMutation(Mutation muta
         // create callback and forward to leader
         logger.trace("Selected {} as leader for mutation with key {}", leader.endpoint(), mutation.key());
 
-        AbstractWriteResponseHandler<Object> handler = strategy.getWriteResponseHandler(plan, null, WriteType.SIMPLE, null, requestTime);
+        AbstractWriteResponseHandler<Object> handler = strategy.getWriteResponseHandler(plan, onComplete, WriteType.SIMPLE, null, requestTime);
 
         // Add callbacks for replicas to respond directly to coordinator
         Message<MutationRequest> toLeader = Message.outWithRequestTime(Verb.FORWARD_WRITE_REQ, new MutationRequest(mutation, plan), requestTime);
@@ -348,6 +356,14 @@ public static AbstractWriteResponseHandler<Object> forwardCounterMutation(Counte
                                                                               ReplicaPlan.ForWrite plan,
                                                                               AbstractReplicationStrategy strategy,
                                                                               Dispatcher.RequestTime requestTime)
+    {
+        return forwardCounterMutationInternal(counterMutation, plan, strategy, requestTime, null);
+    }
+
+    private static AbstractWriteResponseHandler<Object> forwardCounterMutationInternal(
+        CounterMutation counterMutation, ReplicaPlan.ForWrite plan,
+        AbstractReplicationStrategy strategy, Dispatcher.RequestTime requestTime,
+        Consumer<AbstractWriteResponseHandler<?>> onComplete)
     {
         Preconditions.checkArgument(counterMutation.id().isNone(), "CounterMutation should not have an ID when forwarding");
 
@@ -373,7 +389,7 @@ public static AbstractWriteResponseHandler<Object> forwardCounterMutation(Counte
         logger.trace("Forwarding tracked counter mutation to leader replica {}", leader);
 
         // Create response handler for all replicas
-        AbstractWriteResponseHandler<Object> handler = strategy.getWriteResponseHandler(plan, null, WriteType.COUNTER, null, requestTime);
+        AbstractWriteResponseHandler<Object> handler = strategy.getWriteResponseHandler(plan, onComplete, WriteType.COUNTER, null, requestTime);
 
         // Add callbacks for all live replicas to respond directly to coordinator
         Message<CounterMutation> forwardMessage = Message.outWithRequestTime(Verb.COUNTER_MUTATION_REQ, counterMutation, requestTime);
@@ -417,6 +433,33 @@ public static AbstractWriteResponseHandler<Object> forward(IMutation mutation,
             return forwardMutation((Mutation) mutation, plan, strategy, requestTime);
     }
 
+    /**
+     * Forward a mutation to a replica leader for processing.
+     * Dispatches to the appropriate method based on mutation type.
+     *
+     * <p>Like {@link #forward(IMutation, ReplicaPlan.ForWrite, AbstractReplicationStrategy, Dispatcher.RequestTime)},
+     * but wires a completion callback on the handler before any messages are sent, avoiding races where the handler
+     * is signaled before the caller can observe it.
+     *
+     * @param mutation    the mutation to forward (can be Mutation or CounterMutation)
+     * @param plan        the replica plan
+     * @param strategy    the replication strategy
+     * @param requestTime the request time
+     * @param onComplete  callback invoked when the write response handler completes
+     * @return the write response handler
+     */
+    static AbstractWriteResponseHandler<?> forward(IMutation mutation,
+                                                   ReplicaPlan.ForWrite plan,
+                                                   AbstractReplicationStrategy strategy,
+                                                   Dispatcher.RequestTime requestTime,
+                                                   Consumer<AbstractWriteResponseHandler<?>> onComplete)
+    {
+        if (mutation instanceof CounterMutation)
+            return forwardCounterMutationInternal((CounterMutation) mutation, plan, strategy, requestTime, onComplete);
+        else
+            return forwardMutationInternal((Mutation) mutation, plan, strategy, requestTime, onComplete);
+    }
+
     /**
      * Apply a forwarded tracked counter mutation on the leader replica.
      * Called by CounterMutationVerbHandler when receiving a forwarded counter write.