refactor(consensuscommit): unify commit-phase timestamps via orchestrator-provided timestamps

Previously every record/row written during a single ConsensusCommit
transaction generated its own System.currentTimeMillis() at the call site, so
preparedAt/committedAt could differ across rows of the same transaction and
between the data records and the COMMITTED Coordinator state row.

Align timestamps where multiple writes must share one value: the caller
generates the timestamp once and passes it down as an explicit method argument.

- preparedAt: generated once and passed to prepareRecords, so every prepared row
  shares it -- and, with the upcoming TwoPhaseCommit interfaces, so multiple
  participants can be given the same value.
- committedAt: generated once and passed to commitState + commitRecords, so the
  COMMITTED Coordinator state row and every committed data row share it
  (group-commit-disabled).

The no-timestamp constructors of PrepareMutationComposer and CommitMutationComposer
are removed so the caller always supplies the value (RecoveryHandler's
roll-forward passes System.currentTimeMillis() explicitly). AbstractMutationComposer
keeps only the explicit-timestamp constructor and no longer calls
System.currentTimeMillis() itself.

Deliberately NOT threaded, because there is no cross-operation alignment to
enforce -- these generate their own timestamp internally:
- one-phase commit (onePhaseCommitRecords / OnePhaseCommitMutationComposer): the
  single-participant fast path; one composer writes all rows, with no Coordinator
  state row, so its rows already share one timestamp for both PREPARED_AT and
  COMMITTED_AT.
- abort (abortState / abortStateWithoutWriteSet): writes a single Coordinator row.
- rollback (rollbackRecords / RollbackMutationComposer): restores before-images
  and writes no fresh timestamp.
- Coordinator.forceAbort: its parent-id marker and full-id ABORTED rows can be
  written by different lazy-recovery runs, so no invariant binds their createdAt;
  Coordinator.java is left unchanged.
- group-commit Emitter's batched COMMITTED row: a shared row cannot carry a
  per-transaction value, so it keeps an emit-time timestamp.
- RecoveryHandler/RecoveryExecutor: independent transactions, out of scope.

Author: brfrn169

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

@@ -1,6 +1,5 @@
 package com.scalar.db.transaction.consensuscommit;
 
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.scalar.db.api.Mutation;
 import com.scalar.db.api.PutBuilder;
@@ -15,22 +14,14 @@
 public abstract class AbstractMutationComposer implements MutationComposer {
   protected final String id;
   protected final List<Mutation> mutations;
-  protected final long current;
+  protected final long timestamp;
   protected final TransactionTableMetadataManager tableMetadataManager;
 
-  public AbstractMutationComposer(String id, TransactionTableMetadataManager tableMetadataManager) {
+  public AbstractMutationComposer(
+      String id, long timestamp, TransactionTableMetadataManager tableMetadataManager) {
     this.id = id;
     this.mutations = new ArrayList<>();
-    this.current = System.currentTimeMillis();
-    this.tableMetadataManager = tableMetadataManager;
-  }
-
-  @VisibleForTesting
-  AbstractMutationComposer(
-      String id, long current, TransactionTableMetadataManager tableMetadataManager) {
-    this.id = id;
-    this.mutations = new ArrayList<>();
-    this.current = current;
+    this.timestamp = timestamp;
     this.tableMetadataManager = tableMetadataManager;
   }
 

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

@@ -172,7 +172,7 @@ public void commit(TransactionContext context)
 
     if (hasWritesOrDeletesInSnapshot) {
       try {
-        prepareRecords(context);
+        prepareRecords(context, System.currentTimeMillis());
       } catch (PreparationException e) {
         safelyCallOnFailureBeforeCommit(context);
         abortState(context);
@@ -207,8 +207,12 @@ public void commit(TransactionContext context)
         context, beforePreparationHookFuture.orElse(null), hasWritesOrDeletesInSnapshot);
 
     if (hasWritesOrDeletesInSnapshot || !coordinatorWriteOmissionOnReadOnlyEnabled) {
+      long committedAt;
       try {
-        commitState(context);
+        // commitState writes the COMMITTED Coordinator state row and returns the committedAt it
+        // stamped (in group commit, the emit-time value shared across the batch). The data records
+        // are then committed with the same value so the row and the records share one timestamp.
+        committedAt = 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
@@ -219,9 +223,9 @@ public void commit(TransactionContext context)
         }
         throw e;
       }
-    }
-    if (hasWritesOrDeletesInSnapshot) {
-      commitRecords(context);
+      if (hasWritesOrDeletesInSnapshot) {
+        commitRecords(context, committedAt);
+      }
     }
   }
 
@@ -251,16 +255,17 @@ private void abortStateAndRollbackRecordsIfNeeded(
   //
   // TODO: revisit this if/when the Two-phase Commit I/F is removed.
 
-  public void prepareRecords(TransactionContext context) throws PreparationException {
-    participantCommitHandler.prepareRecords(context);
+  public void prepareRecords(TransactionContext context, long preparedAt)
+      throws PreparationException {
+    participantCommitHandler.prepareRecords(context, preparedAt);
   }
 
   public void validateRecords(TransactionContext context) throws ValidationException {
     participantCommitHandler.validateRecords(context);
   }
 
-  public void commitRecords(TransactionContext context) {
-    participantCommitHandler.commitRecords(context);
+  public void commitRecords(TransactionContext context, long committedAt) {
+    participantCommitHandler.commitRecords(context, committedAt);
   }
 
   public void rollbackRecords(TransactionContext context) {
@@ -285,14 +290,14 @@ void onePhaseCommitRecords(TransactionContext context)
     participantCommitHandler.onePhaseCommitRecords(context);
   }
 
-  public void commitState(TransactionContext context)
+  public long commitState(TransactionContext context)
       throws CommitConflictException, UnknownTransactionStatusException {
-    coordinatorCommitHandler.commitState(context);
+    return coordinatorCommitHandler.commitState(context);
   }
 
-  public void commitStateWithoutWriteSet(TransactionContext context)
+  public long commitStateWithoutWriteSet(TransactionContext context)
       throws CommitConflictException, UnknownTransactionStatusException {
-    coordinatorCommitHandler.commitStateWithoutWriteSet(context);
+    return coordinatorCommitHandler.commitStateWithoutWriteSet(context);
   }
 
   public TransactionState abortState(TransactionContext context)

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

@@ -5,7 +5,6 @@
 import static com.scalar.db.transaction.consensuscommit.Attribute.STATE;
 import static com.scalar.db.transaction.consensuscommit.ConsensusCommitUtils.getTransactionTableMetadata;
 
-import com.google.common.annotations.VisibleForTesting;
 import com.scalar.db.api.ConditionBuilder;
 import com.scalar.db.api.Consistency;
 import com.scalar.db.api.Delete;
@@ -31,14 +30,9 @@
 public class CommitMutationComposer extends AbstractMutationComposer {
   private static final Logger logger = LoggerFactory.getLogger(CommitMutationComposer.class);
 
-  public CommitMutationComposer(String id, TransactionTableMetadataManager tableMetadataManager) {
-    super(id, tableMetadataManager);
-  }
-
-  @VisibleForTesting
-  CommitMutationComposer(
-      String id, long current, TransactionTableMetadataManager tableMetadataManager) {
-    super(id, current, tableMetadataManager);
+  public CommitMutationComposer(
+      String id, long timestamp, TransactionTableMetadataManager tableMetadataManager) {
+    super(id, timestamp, tableMetadataManager);
   }
 
   @Override
@@ -95,7 +89,7 @@ private Put composePut(Operation base, @Nullable TransactionResult result)
                         ConditionBuilder.column(STATE)
                             .isEqualToInt(TransactionState.PREPARED.get()))
                     .build())
-            .bigIntValue(COMMITTED_AT, current)
+            .bigIntValue(COMMITTED_AT, timestamp)
             .intValue(STATE, TransactionState.COMMITTED.get())
             .consistency(Consistency.LINEARIZABLE);
     getClusteringKey(base, result).ifPresent(putBuilder::clusteringKey);

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

@@ -40,30 +40,38 @@ class CoordinatorCommitHandler {
   /**
    * Writes the COMMITTED state with the {@code tx_write_set} populated from the given context's
    * snapshot.
+   *
+   * @return the {@code committedAt} timestamp written to the COMMITTED Coordinator state row. The
+   *     caller stamps the committed data records with the same value so the row and the records
+   *     share one timestamp.
    */
-  void commitState(TransactionContext context)
+  long commitState(TransactionContext context)
       throws CommitConflictException, UnknownTransactionStatusException {
-    commitStateInternal(context, writeSetEncoder.encodeSingleGroupWriteSet(context, false));
+    return commitStateInternal(context, writeSetEncoder.encodeSingleGroupWriteSet(context, false));
   }
 
-  /** Writes the COMMITTED state without persisting a {@code tx_write_set}. */
-  void commitStateWithoutWriteSet(TransactionContext context)
+  /**
+   * Writes the COMMITTED state without persisting a {@code tx_write_set}.
+   *
+   * @return the {@code committedAt} timestamp written to the COMMITTED Coordinator state row.
+   */
+  long commitStateWithoutWriteSet(TransactionContext context)
       throws CommitConflictException, UnknownTransactionStatusException {
-    commitStateInternal(context, null);
+    return commitStateInternal(context, null);
   }
 
-  private void commitStateInternal(TransactionContext context, @Nullable WriteSet writeSet)
+  private long commitStateInternal(TransactionContext context, @Nullable WriteSet writeSet)
       throws CommitConflictException, UnknownTransactionStatusException {
     String id = context.transactionId;
     try {
+      long committedAt = System.currentTimeMillis();
       Coordinator.State state =
-          new Coordinator.State(
-              id, writeSet, TransactionState.COMMITTED, System.currentTimeMillis());
+          new Coordinator.State(id, writeSet, TransactionState.COMMITTED, committedAt);
       coordinator.putState(state);
-      logger.debug(
-          "Transaction {} is committed successfully at {}", id, System.currentTimeMillis());
+      logger.debug("Transaction {} is committed successfully at {}", id, committedAt);
+      return committedAt;
     } catch (CoordinatorConflictException e) {
-      handleCommitConflict(context, e);
+      return handleCommitConflict(context, e);
     } catch (CoordinatorException e) {
       throw new UnknownTransactionStatusException(
           CoreError.CONSENSUS_COMMIT_UNKNOWN_COORDINATOR_STATUS.buildMessage(e.getMessage()),
@@ -72,13 +80,18 @@ private void commitStateInternal(TransactionContext context, @Nullable WriteSet
     }
   }
 
-  void handleCommitConflict(TransactionContext context, Exception cause)
+  /**
+   * Resolves a putState conflict. Returns the {@code committedAt} of the already-persisted
+   * COMMITTED state when this transaction turns out to be already committed; otherwise rolls back
+   * the records and throws.
+   */
+  long handleCommitConflict(TransactionContext context, Exception cause)
       throws CommitConflictException, UnknownTransactionStatusException {
     try {
       Optional<Coordinator.State> s = coordinator.getState(context.transactionId);
       if (s.isPresent()) {
-        TransactionState state = s.get().getState();
-        if (state == TransactionState.ABORTED) {
+        Coordinator.State persisted = s.get();
+        if (persisted.getState() == TransactionState.ABORTED) {
           throw new CommitConflictException(
               CoreError.CONSENSUS_COMMIT_CONFLICT_OCCURRED_WHEN_COMMITTING_STATE.buildMessage(
                   cause.getMessage()),
@@ -93,10 +106,13 @@ void handleCommitConflict(TransactionContext context, Exception cause)
         // putIfNotExists race and observes it here. With One-phase Commit I/F this is unreachable:
         // this commit is the only writer of the COMMITTED state and it just lost the race, so the
         // conflicting row was an ABORTED from a lazy recovery, handled above. The transaction is
-        // committed, so return normally and let the caller commit the records.
+        // committed, so return the persisted row's committedAt and let the caller commit the
+        // records with it (keeping the row and the records on a single timestamp).
         //
         // TODO: revisit this if/when the Two-phase Commit I/F is removed -- it would then be
         // unreachable (a COMMITTED state could never be observed after a conflict here).
+
+        return persisted.getCreatedAt();
       } else {
         // 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

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

@@ -44,23 +44,32 @@ class CoordinatorCommitHandlerWithGroupCommit extends CoordinatorCommitHandler {
     this.groupCommitter = groupCommitter;
   }
 
+  /**
+   * Group commits the COMMITTED state. The {@code committedAt} stamped on the Coordinator row is
+   * generated by the Emitter when it writes the batched row, and that single value is handed back
+   * to every transaction in the batch through {@link CoordinatorGroupCommitter#ready}. This method
+   * returns it so the caller can stamp this transaction's committed data records with the same
+   * value, keeping the (batched) Coordinator row and the records on a single timestamp.
+   */
   @Override
-  void commitState(TransactionContext context)
+  long commitState(TransactionContext context)
       throws CommitConflictException, UnknownTransactionStatusException {
     String id = context.transactionId;
     try {
-      // Group commit the state by internally calling `groupCommitState()` via the emitter.
-      groupCommitter.ready(id, context);
-      logger.debug(
-          "Transaction {} is committed successfully at {}", id, System.currentTimeMillis());
+      // Group commit the state by internally calling `groupCommitState()` via the emitter. The
+      // returned value is the emit-time committedAt the Emitter wrote to the batched row.
+      Long committedAt = groupCommitter.ready(id, context);
+      assert committedAt != null;
+      logger.debug("Transaction {} is committed successfully at {}", id, committedAt);
+      return committedAt;
     } catch (GroupCommitConflictException e) {
       cancelGroupCommitIfNeeded(context);
-      handleCommitConflict(context, e);
+      return handleCommitConflict(context, e);
     } catch (GroupCommitException e) {
       cancelGroupCommitIfNeeded(context);
       Throwable cause = e.getCause();
       if (cause instanceof CoordinatorConflictException) {
-        handleCommitConflict(context, (CoordinatorConflictException) cause);
+        return handleCommitConflict(context, (CoordinatorConflictException) cause);
       } else {
         // Failed to access the coordinator state. The state is unknown.
         throw new UnknownTransactionStatusException("Coordinator status is unknown", cause, id);
@@ -112,7 +121,7 @@ void cancelGroupCommitIfNeeded(TransactionContext context) {
   }
 
   @VisibleForTesting
-  static class Emitter implements Emittable<String, String, TransactionContext, Void> {
+  static class Emitter implements Emittable<String, String, TransactionContext, Long> {
     private final Coordinator coordinator;
     private final WriteSetEncoder writeSetEncoder;
 
@@ -122,7 +131,7 @@ static class Emitter implements Emittable<String, String, TransactionContext, Vo
     }
 
     @Override
-    public Void emitNormalGroup(String parentId, List<TransactionContext> contexts)
+    public Long emitNormalGroup(String parentId, List<TransactionContext> contexts)
         throws CoordinatorException {
       if (contexts.isEmpty()) {
         return null;
@@ -135,32 +144,33 @@ public Void emitNormalGroup(String parentId, List<TransactionContext> contexts)
       for (TransactionContext context : contexts) {
         childIds.add(KEY_MANIPULATOR.keysFromFullKey(context.transactionId).childKey);
       }
+      // Generate one committedAt for the whole batched row and return it so every transaction in
+      // this group stamps its committed data records with the same value.
+      long committedAt = System.currentTimeMillis();
       coordinator.putState(
           new State(
               parentId,
               childIds,
               writeSetEncoder.encodeMultiGroupWriteSet(contexts, false),
               TransactionState.COMMITTED,
-              System.currentTimeMillis()));
+              committedAt));
       logger.debug(
-          "Transaction {} (parent ID) is committed successfully at {}",
-          parentId,
-          System.currentTimeMillis());
-      return null;
+          "Transaction {} (parent ID) is committed successfully at {}", parentId, committedAt);
+      return committedAt;
     }
 
     @Override
-    public Void emitDelayedGroup(String fullId, TransactionContext context)
+    public Long emitDelayedGroup(String fullId, TransactionContext context)
         throws CoordinatorException {
+      long committedAt = System.currentTimeMillis();
       coordinator.putState(
           new State(
               fullId,
               writeSetEncoder.encodeSingleGroupWriteSet(context, false),
               TransactionState.COMMITTED,
-              System.currentTimeMillis()));
-      logger.debug(
-          "Transaction {} is committed successfully at {}", fullId, System.currentTimeMillis());
-      return null;
+              committedAt));
+      logger.debug("Transaction {} is committed successfully at {}", fullId, committedAt);
+      return committedAt;
     }
   }
 }

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

@@ -6,7 +6,7 @@
 import java.util.Optional;
 
 public class CoordinatorGroupCommitter
-    extends GroupCommitter<String, String, String, String, String, TransactionContext, Void> {
+    extends GroupCommitter<String, String, String, String, String, TransactionContext, Long> {
   CoordinatorGroupCommitter(GroupCommitConfig config) {
     super("coordinator", config, new CoordinatorGroupCommitKeyManipulator());
   }

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

@@ -23,15 +23,18 @@
 @NotThreadSafe
 public class OnePhaseCommitMutationComposer extends AbstractMutationComposer {
 
+  // One-phase commit is the single-participant fast path, so there is no cross-participant
+  // timestamp to coordinate; the composer generates its own timestamp for both PREPARED_AT and
+  // COMMITTED_AT.
   public OnePhaseCommitMutationComposer(
       String id, TransactionTableMetadataManager tableMetadataManager) {
-    super(id, tableMetadataManager);
+    super(id, System.currentTimeMillis(), tableMetadataManager);
   }
 
   @VisibleForTesting
   OnePhaseCommitMutationComposer(
-      String id, long current, TransactionTableMetadataManager tableMetadataManager) {
-    super(id, current, tableMetadataManager);
+      String id, long timestamp, TransactionTableMetadataManager tableMetadataManager) {
+    super(id, timestamp, tableMetadataManager);
   }
 
   @Override
@@ -66,8 +69,8 @@ private Put composePut(Put base, @Nullable TransactionResult result) throws Exec
 
     putBuilder.textValue(Attribute.ID, id);
     putBuilder.intValue(Attribute.STATE, TransactionState.COMMITTED.get());
-    putBuilder.bigIntValue(Attribute.PREPARED_AT, current);
-    putBuilder.bigIntValue(Attribute.COMMITTED_AT, current);
+    putBuilder.bigIntValue(Attribute.PREPARED_AT, timestamp);
+    putBuilder.bigIntValue(Attribute.COMMITTED_AT, timestamp);
 
     if (!isInsertModeEnabled(base) && result != null) { // overwrite existing record
       int version = result.getVersion();