refactor(consensuscommit): decouple CoordinatorCommitHandler from ParticipantCommitHandler

CoordinatorCommitHandler held a ParticipantCommitHandler purely to roll back
the transaction's prepared records when a commitState putState race resolved
to ABORTED/absent. That made the Coordinator-side (state-table) handler depend
on the participant-side (data-record) handler, which blocks reusing the two
handlers independently (e.g. from callers other than CommitHandler).

Move that responsibility to the orchestrator: handleCommitConflict now only
reports the conflict (throws CommitConflictException, or returns normally when
the state is already COMMITTED), and CommitHandler.commit() catches it and
rolls back the records via its own ParticipantCommitHandler. No behavioral
change.

- Drop the ParticipantCommitHandler dependency from CoordinatorCommitHandler
  and CoordinatorCommitHandlerWithGroupCommit.
- Order the two handlers Coordinator-before-Participant in CommitHandler's
  fields/constructors now that they are independent.
- Drop the "one of the two specialized handlers that CommitHandler delegates
  to" framing from the handler Javadocs; describe each handler by its own
  responsibility so it reads correctly for non-CommitHandler callers.

Author: brfrn169

core/src/main/java/com/scalar/db/transaction/consensuscommit/CommitHandler.java

@@ -49,8 +49,8 @@
 public class CommitHandler {
   private static final Logger logger = LoggerFactory.getLogger(CommitHandler.class);
 
-  private final ParticipantCommitHandler participantCommitHandler;
   private final CoordinatorCommitHandler coordinatorCommitHandler;
+  private final ParticipantCommitHandler participantCommitHandler;
   protected final boolean coordinatorWriteOmissionOnReadOnlyEnabled;
 
   @LazyInit @Nullable private BeforePreparationHook beforePreparationHook;
@@ -65,16 +65,15 @@ public CommitHandler(
       boolean coordinatorWriteOmissionOnReadOnlyEnabled,
       boolean onePhaseCommitEnabled) {
     this.coordinatorWriteOmissionOnReadOnlyEnabled = coordinatorWriteOmissionOnReadOnlyEnabled;
+    this.coordinatorCommitHandler =
+        new CoordinatorCommitHandler(coordinator, new WriteSetEncoder(tableMetadataManager));
     this.participantCommitHandler =
         new ParticipantCommitHandler(
             storage,
             tableMetadataManager,
             parallelExecutor,
             mutationsGrouper,
             onePhaseCommitEnabled);
-    this.coordinatorCommitHandler =
-        new CoordinatorCommitHandler(
-            coordinator, new WriteSetEncoder(tableMetadataManager), participantCommitHandler);
   }
 
   // Constructor for subclasses (CommitHandlerWithGroupCommit) that need to inject a
@@ -84,11 +83,11 @@ public CommitHandler(
   @SuppressFBWarnings("EI_EXPOSE_REP2")
   protected CommitHandler(
       boolean coordinatorWriteOmissionOnReadOnlyEnabled,
-      ParticipantCommitHandler participantCommitHandler,
-      CoordinatorCommitHandler coordinatorCommitHandler) {
+      CoordinatorCommitHandler coordinatorCommitHandler,
+      ParticipantCommitHandler participantCommitHandler) {
     this.coordinatorWriteOmissionOnReadOnlyEnabled = coordinatorWriteOmissionOnReadOnlyEnabled;
-    this.participantCommitHandler = checkNotNull(participantCommitHandler);
     this.coordinatorCommitHandler = checkNotNull(coordinatorCommitHandler);
+    this.participantCommitHandler = checkNotNull(participantCommitHandler);
   }
 
   /**
@@ -208,7 +207,18 @@ public void commit(TransactionContext context)
         context, beforePreparationHookFuture.orElse(null), hasWritesOrDeletesInSnapshot);
 
     if (hasWritesOrDeletesInSnapshot || !coordinatorWriteOmissionOnReadOnlyEnabled) {
-      commitState(context);
+      try {
+        commitState(context);
+      } catch (CommitConflictException e) {
+        // The COMMITTED-state write lost a putState race that resolved to ABORTED (or absent): the
+        // transaction is aborted. The Coordinator-side handler only reports the conflict; the
+        // orchestrator owns the records, so roll back the prepared records here before surfacing
+        // it.
+        if (hasWritesOrDeletesInSnapshot) {
+          rollbackRecords(context);
+        }
+        throw e;
+      }
     }
     if (hasWritesOrDeletesInSnapshot) {
       commitRecords(context);

core/src/main/java/com/scalar/db/transaction/consensuscommit/CommitHandlerWithGroupCommit.java

@@ -46,8 +46,8 @@ public CommitHandlerWithGroupCommit(
         checkNotNull(groupCommitter));
   }
 
-  // Second-stage delegating constructor so the ParticipantCommitHandler is available when we build
-  // the CoordinatorCommitHandlerWithGroupCommit.
+  // Second-stage delegating constructor that builds the two specialized handlers before handing
+  // them to the package-private constructor.
   private CommitHandlerWithGroupCommit(
       TransactionTableMetadataManager tableMetadataManager,
       boolean coordinatorWriteOmissionOnReadOnlyEnabled,
@@ -56,22 +56,19 @@ private CommitHandlerWithGroupCommit(
       CoordinatorGroupCommitter groupCommitter) {
     this(
         coordinatorWriteOmissionOnReadOnlyEnabled,
-        participantCommitHandler,
         new CoordinatorCommitHandlerWithGroupCommit(
-            coordinator,
-            new WriteSetEncoder(tableMetadataManager),
-            participantCommitHandler,
-            groupCommitter));
+            coordinator, new WriteSetEncoder(tableMetadataManager), groupCommitter),
+        participantCommitHandler);
   }
 
-  // Package-private so test code can inject Mockito spies of ParticipantCommitHandler /
-  // CoordinatorCommitHandlerWithGroupCommit via constructor injection rather than via reflection.
+  // Package-private so test code can inject Mockito spies of the two specialized handlers via
+  // constructor injection rather than via reflection.
   @SuppressFBWarnings("EI_EXPOSE_REP2")
   CommitHandlerWithGroupCommit(
       boolean coordinatorWriteOmissionOnReadOnlyEnabled,
-      ParticipantCommitHandler participantCommitHandler,
-      CoordinatorCommitHandlerWithGroupCommit coordinatorHandler) {
-    super(coordinatorWriteOmissionOnReadOnlyEnabled, participantCommitHandler, coordinatorHandler);
+      CoordinatorCommitHandlerWithGroupCommit coordinatorHandler,
+      ParticipantCommitHandler participantCommitHandler) {
+    super(coordinatorWriteOmissionOnReadOnlyEnabled, coordinatorHandler, participantCommitHandler);
     this.coordinatorHandler = coordinatorHandler;
   }
 

core/src/main/java/com/scalar/db/transaction/consensuscommit/CoordinatorCommitHandler.java

@@ -18,32 +18,23 @@
  * Handles the Coordinator-side (state-table) operations of the Consensus Commit protocol: writing
  * the COMMITTED / ABORTED records to the Coordinator state table and resolving putState conflicts.
  *
- * <p>This is one of the two specialized handlers that {@link CommitHandler} delegates to. The other
- * is {@link ParticipantCommitHandler}, which owns participant-side data-record operations.
- *
- * <p>Methods here only touch the Coordinator state table via {@link Coordinator}. They never write
- * to user data tables — except indirectly: when {@code commitState} loses a putState race and the
- * persisted state turns out to be ABORTED (or absent), this handler asks the orchestrator-provided
- * {@link ParticipantCommitHandler} to roll back the records that the transaction had prepared. That
- * cross-handler call keeps the "commit conflict → rollback records" invariant atomic from the
- * orchestrator's perspective.
+ * <p>Methods here only touch the Coordinator state table via {@link Coordinator}; they never touch
+ * user data tables. When {@code commitState} loses a putState race and the persisted state turns
+ * out to be ABORTED (or absent), this handler reports a {@link CommitConflictException} and leaves
+ * the rollback of the transaction's prepared records to the caller. This keeps the handler free of
+ * any participant-side dependency.
  */
 @ThreadSafe
 class CoordinatorCommitHandler {
   private static final Logger logger = LoggerFactory.getLogger(CoordinatorCommitHandler.class);
 
   private final Coordinator coordinator;
   private final WriteSetEncoder writeSetEncoder;
-  private final ParticipantCommitHandler participantCommitHandler;
 
   @SuppressFBWarnings("EI_EXPOSE_REP2")
-  CoordinatorCommitHandler(
-      Coordinator coordinator,
-      WriteSetEncoder writeSetEncoder,
-      ParticipantCommitHandler participantCommitHandler) {
+  CoordinatorCommitHandler(Coordinator coordinator, WriteSetEncoder writeSetEncoder) {
     this.coordinator = checkNotNull(coordinator);
     this.writeSetEncoder = checkNotNull(writeSetEncoder);
-    this.participantCommitHandler = checkNotNull(participantCommitHandler);
   }
 
   /**
@@ -88,7 +79,6 @@ void handleCommitConflict(TransactionContext context, Exception cause)
       if (s.isPresent()) {
         TransactionState state = s.get().getState();
         if (state == TransactionState.ABORTED) {
-          participantCommitHandler.rollbackRecords(context);
           throw new CommitConflictException(
               CoreError.CONSENSUS_COMMIT_CONFLICT_OCCURRED_WHEN_COMMITTING_STATE.buildMessage(
                   cause.getMessage()),
@@ -111,8 +101,9 @@ void handleCommitConflict(TransactionContext context, Exception cause)
         // The coordinator state is absent: a row existed when our putIfNotExists lost the race, but
         // it is gone now. In both interfaces this means the conflicting row was an ABORTED written
         // by a lazy recovery (which also rolled the records back) and later removed by the
-        // Coordinator state cleanup process, so the transaction is definitively aborted. Roll the
-        // records back and report a conflict -- the same outcome as the present-ABORTED case above.
+        // Coordinator state cleanup process, so the transaction is definitively aborted. Report a
+        // conflict (the orchestrator rolls the records back) -- the same outcome as the
+        // present-ABORTED case above.
         //
         // A COMMITTED row can be ruled out here in both interfaces:
         //   - One-phase Commit I/F: this commit is the only writer of this transaction's COMMITTED
@@ -139,7 +130,6 @@ void handleCommitConflict(TransactionContext context, Exception cause)
         //
         // TODO: revisit this if/when the Two-phase Commit I/F is removed.
 
-        participantCommitHandler.rollbackRecords(context);
         throw new CommitConflictException(
             CoreError.CONSENSUS_COMMIT_CONFLICT_OCCURRED_WHEN_COMMITTING_STATE.buildMessage(
                 cause.getMessage()),

core/src/main/java/com/scalar/db/transaction/consensuscommit/CoordinatorCommitHandlerWithGroupCommit.java

@@ -22,9 +22,6 @@
  * Group-commit-aware variant of {@link CoordinatorCommitHandler}. Routes {@code commitState} into
  * the group-commit emitter so that multiple transactions' COMMITTED states can be batched into one
  * Coordinator-table row. Cancels the group commit slot reservation on the abort path.
- *
- * <p>Used by {@link CommitHandlerWithGroupCommit} (the orchestrator variant) in place of the base
- * {@link CoordinatorCommitHandler} when group commit is enabled.
  */
 @ThreadSafe
 class CoordinatorCommitHandlerWithGroupCommit extends CoordinatorCommitHandler {
@@ -39,9 +36,8 @@ class CoordinatorCommitHandlerWithGroupCommit extends CoordinatorCommitHandler {
   CoordinatorCommitHandlerWithGroupCommit(
       Coordinator coordinator,
       WriteSetEncoder writeSetEncoder,
-      ParticipantCommitHandler participantCommitHandler,
       CoordinatorGroupCommitter groupCommitter) {
-    super(coordinator, writeSetEncoder, participantCommitHandler);
+    super(coordinator, writeSetEncoder);
     checkNotNull(groupCommitter);
     // The methods of this emitter will be called via GroupCommitter.ready().
     groupCommitter.setEmitter(new Emitter(coordinator, writeSetEncoder));

core/src/main/java/com/scalar/db/transaction/consensuscommit/ParticipantCommitHandler.java

@@ -33,9 +33,6 @@
  * preparing, validating, committing, and rolling back the records the transaction touches in user
  * data tables.
  *
- * <p>This is one of the two specialized handlers that {@link CommitHandler} delegates to. The other
- * is {@link CoordinatorCommitHandler}, which owns Coordinator-table state writes.
- *
  * <p>Methods here only touch user data tables via {@link DistributedStorage} and the {@link
  * MutationsGrouper} / {@link ParallelExecutor} stack. They never write to the Coordinator table.
  */

core/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommit.java

@@ -234,8 +234,13 @@ public void commit() throws CommitConflictException, UnknownTransactionStatusExc
 
     try {
       commit.commitStateWithoutWriteSet(context);
-    } catch (CommitConflictException | UnknownTransactionStatusException e) {
-      // no need to rollback because the transaction has already been rolled back
+    } catch (CommitConflictException e) {
+      commit.rollbackRecords(context);
+      needRollback = false;
+      throw e;
+    } catch (UnknownTransactionStatusException e) {
+      // The coordinator state is unknown, so we must not roll back the records: the transaction may
+      // have committed.
       needRollback = false;
 
       throw e;

core/src/test/java/com/scalar/db/transaction/consensuscommit/CommitHandlerTest.java

@@ -68,8 +68,8 @@ protected void extraCleanup() {}
   protected CommitHandler createCommitHandler(boolean coordinatorWriteOmissionOnReadOnlyEnabled) {
     return new CommitHandler(
         coordinatorWriteOmissionOnReadOnlyEnabled,
-        participantCommitHandler,
-        coordinatorCommitHandler);
+        coordinatorCommitHandler,
+        participantCommitHandler);
   }
 
   @BeforeEach
@@ -306,6 +306,29 @@ public void commit_ValidateThrowsGenericException_WithWrites_ShouldDelegateAbort
     verify(participantCommitHandler).rollbackRecords(context);
   }
 
+  @Test
+  public void commit_CommitStateThrowsConflict_WithWrites_ShouldRollbackRecordsAndThrowConflict()
+      throws Exception {
+    // The Coordinator-side handler only reports the commit-state putState conflict; the
+    // orchestrator
+    // owns the records, so it rolls them back here before surfacing the conflict.
+
+    // Arrange
+    Snapshot snapshot = snapshotWithWrites();
+    doThrow(new CommitConflictException("conflict", anyId()))
+        .when(coordinatorCommitHandler)
+        .commitState(any());
+    TransactionContext context =
+        createTransactionContext(anyId(), snapshot, Isolation.SNAPSHOT, false, false);
+
+    // Act Assert
+    assertThatThrownBy(() -> handler.commit(context)).isInstanceOf(CommitConflictException.class);
+
+    verify(coordinatorCommitHandler).commitState(context);
+    verify(participantCommitHandler).rollbackRecords(context);
+    verify(participantCommitHandler, never()).commitRecords(any());
+  }
+
   @Test
   public void commit_ValidateThrows_NoWrites_OmissionEnabled_ShouldNotDelegateAbortNorRollback()
       throws Exception {

core/src/test/java/com/scalar/db/transaction/consensuscommit/CommitHandlerWithGroupCommitTest.java

@@ -46,8 +46,8 @@ protected CommitHandler createCommitHandler(boolean coordinatorWriteOmissionOnRe
     coordinatorCommitHandler = groupCommitCoordinatorHandler;
     return new CommitHandlerWithGroupCommit(
         coordinatorWriteOmissionOnReadOnlyEnabled,
-        participantCommitHandler,
-        groupCommitCoordinatorHandler);
+        groupCommitCoordinatorHandler,
+        participantCommitHandler);
   }
 
   // =========================================================================

core/src/test/java/com/scalar/db/transaction/consensuscommit/CoordinatorCommitHandlerTest.java

@@ -40,7 +40,6 @@ class CoordinatorCommitHandlerTest {
 
   @Mock private Coordinator coordinator;
   @Mock private TransactionTableMetadataManager tableMetadataManager;
-  @Mock private ParticipantCommitHandler participantCommitHandler;
   @Mock private ConsensusCommitConfig config;
 
   private ParallelExecutor parallelExecutor;
@@ -63,8 +62,7 @@ void tearDown() {
   }
 
   private CoordinatorCommitHandler createHandler() {
-    return new CoordinatorCommitHandler(
-        coordinator, new WriteSetEncoder(tableMetadataManager), participantCommitHandler);
+    return new CoordinatorCommitHandler(coordinator, new WriteSetEncoder(tableMetadataManager));
   }
 
   private Put preparePut1() {
@@ -127,9 +125,11 @@ void commitState_WhenSuccessful_ShouldPutCommittedState()
   }
 
   @Test
-  void
-      commitState_WhenCoordinatorConflictAndAbortedReturnedInGetState_ShouldRollbackRecordsAndThrowConflict()
-          throws Exception {
+  void commitState_WhenCoordinatorConflictAndAbortedReturnedInGetState_ShouldThrowConflict()
+      throws Exception {
+    // Record rollback is the orchestrator's responsibility (see CommitHandlerTest); here we only
+    // assert that the conflict is surfaced.
+
     // Arrange
     Snapshot snapshot = prepareSnapshotWithWrite();
     TransactionContext context = createTransactionContext(snapshot);
@@ -149,8 +149,6 @@ void commitState_WhenSuccessful_ShouldPutCommittedState()
     // Act Assert
     assertThatThrownBy(() -> handler.commitState(context))
         .isInstanceOf(CommitConflictException.class);
-
-    verify(participantCommitHandler).rollbackRecords(context);
   }
 
   @Test
@@ -175,17 +173,16 @@ void commitState_WhenCoordinatorConflictAndCommittedReturnedInGetState_ShouldRet
         .when(coordinator)
         .getState(anyId());
 
-    // Act (must not throw)
+    // Act Assert (must not throw — the transaction is already committed)
     handler.commitState(context);
-
-    // Assert
-    verify(participantCommitHandler, never()).rollbackRecords(any());
   }
 
   @Test
-  void
-      commitState_WhenCoordinatorConflictAndNoStatePersisted_ShouldRollbackRecordsAndThrowConflict()
-          throws Exception {
+  void commitState_WhenCoordinatorConflictAndNoStatePersisted_ShouldThrowConflict()
+      throws Exception {
+    // Record rollback is the orchestrator's responsibility (see CommitHandlerTest); here we only
+    // assert that the conflict is surfaced.
+
     // Arrange
     Snapshot snapshot = prepareSnapshotWithWrite();
     TransactionContext context = createTransactionContext(snapshot);
@@ -200,8 +197,6 @@ void commitState_WhenCoordinatorConflictAndCommittedReturnedInGetState_ShouldRet
     // Act Assert
     assertThatThrownBy(() -> handler.commitState(context))
         .isInstanceOf(CommitConflictException.class);
-
-    verify(participantCommitHandler).rollbackRecords(context);
   }
 
   @Test
@@ -435,8 +430,10 @@ void forceAbortState_WhenCoordinatorExceptionThrown_ShouldThrowUnknown()
   // ---------- handleCommitConflict ----------
 
   @Test
-  void handleCommitConflict_WhenAbortedStatePersisted_ShouldRollbackRecordsAndThrowConflict()
-      throws Exception {
+  void handleCommitConflict_WhenAbortedStatePersisted_ShouldThrowConflict() throws Exception {
+    // Record rollback is the orchestrator's responsibility (see CommitHandlerTest); here we only
+    // assert that the conflict is surfaced.
+
     // Arrange
     Snapshot snapshot = prepareSnapshotWithWrite();
     TransactionContext context = createTransactionContext(snapshot);
@@ -452,7 +449,6 @@ void handleCommitConflict_WhenAbortedStatePersisted_ShouldRollbackRecordsAndThro
     assertThatThrownBy(() -> handler.handleCommitConflict(context, cause))
         .isInstanceOf(CommitConflictException.class)
         .hasCause(cause);
-    verify(participantCommitHandler).rollbackRecords(context);
   }
 
   @Test
@@ -467,16 +463,15 @@ void handleCommitConflict_WhenCommittedStatePersisted_ShouldReturnNormally() thr
         .when(coordinator)
         .getState(anyId());
 
-    // Act (must not throw)
+    // Act Assert (must not throw)
     handler.handleCommitConflict(context, new RuntimeException("conflict"));
-
-    // Assert
-    verify(participantCommitHandler, never()).rollbackRecords(any());
   }
 
   @Test
-  void handleCommitConflict_WhenNoStatePersisted_ShouldRollbackRecordsAndThrowConflict()
-      throws Exception {
+  void handleCommitConflict_WhenNoStatePersisted_ShouldThrowConflict() throws Exception {
+    // Record rollback is the orchestrator's responsibility (see CommitHandlerTest); here we only
+    // assert that the conflict is surfaced.
+
     // Arrange
     Snapshot snapshot = prepareSnapshotWithWrite();
     TransactionContext context = createTransactionContext(snapshot);
@@ -487,7 +482,6 @@ void handleCommitConflict_WhenNoStatePersisted_ShouldRollbackRecordsAndThrowConf
     assertThatThrownBy(() -> handler.handleCommitConflict(context, cause))
         .isInstanceOf(CommitConflictException.class)
         .hasCause(cause);
-    verify(participantCommitHandler).rollbackRecords(context);
   }
 
   @Test