Prohibit rollback of a procedure which has passed its commit point

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -1430,4 +1430,11 @@ private ProcedureMessages() {}
       "DataNode setSystemStatus RPC failed: %s: %s: %s; ";
   public static final String EXCEPTION_CONFIGNODE_TRYSETNODESTATUS_FAILED_ARG_ARG_ARG_C9BA78F4 =
       "ConfigNode trySetNodeStatus failed: %s: %s: %s; ";
+  public static final String
+      LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9 =
+          "Procedure {} is past its commit point; keeping every state it executed.";
+  public static final String LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354 =
+      "finish procedure without rollback";
+  public static final String LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1 =
+      "delete child procedures without rollback";
 }

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -1515,4 +1515,10 @@ private ProcedureMessages() {}
       "DataNode setSystemStatus RPC 失败：%s：%s：%s； ";
   public static final String EXCEPTION_CONFIGNODE_TRYSETNODESTATUS_FAILED_ARG_ARG_ARG_C9BA78F4 =
       "ConfigNode trySetNodeStatus 失败：%s：%s：%s； ";
+  public static final String
+      LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9 =
+          "流程 {} 已越过提交点，保留它执行过的所有状态。";
+  public static final String LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354 = "结束流程（不回滚）";
+  public static final String LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1 =
+      "删除子流程（不回滚）";
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java

@@ -99,6 +99,24 @@ public final boolean hasLock() {
   protected abstract void rollback(Env env)
       throws IOException, InterruptedException, ProcedureException;
 
+  /**
+   * Whether the states this procedure has already executed may still be undone. The executor asks
+   * this once, before it starts to roll back the root procedure stack. A procedure which returns
+   * false is finished without undoing any of its states, see {@link #abandonRollback()}.
+   */
+  protected boolean isRollbackSupported() {
+    return true;
+  }
+
+  /**
+   * Drops the bookkeeping of the states this procedure has executed. The persisted stack indexes
+   * are what a recovered rollback rebuilds its stack from, so dropping them is what makes the
+   * decision survive a leader switch.
+   */
+  protected synchronized void abandonRollback() {
+    stackIndexes = null;
+  }
+
   public void serialize(DataOutputStream stream) throws IOException {
     // procid
     stream.writeLong(this.procId);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java

@@ -40,8 +40,10 @@
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Deque;
 import java.util.HashSet;
+import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -370,7 +372,7 @@ public boolean removeInternalProcedure(InternalProcedure internalProcedure) {
    *
    * @param proc procedure
    */
-  private void executeProcedure(Procedure<Env> proc) {
+  private void executeProcedure(Procedure<Env> proc) throws InterruptedException {
     if (proc.isFinished()) {
       LOG.debug(ProcedureMessages.IS_ALREADY_FINISHED, proc);
       return;
@@ -689,7 +691,7 @@ private void yieldProcedure(Procedure<Env> proc) {
    * @return lock state
    */
   private LockStateResult<Env> executeRootStackRollback(
-      Long rootProcId, RootProcedureStack<Env> procedureStack) {
+      Long rootProcId, RootProcedureStack<Env> procedureStack) throws InterruptedException {
     Procedure<Env> rootProcedure = procedures.get(rootProcId);
     ProcedureException exception = rootProcedure.getException();
     if (exception == null) {
@@ -700,14 +702,33 @@ private LockStateResult<Env> executeRootStackRollback(
           () -> store.update(rootProcedure), "root procedure rollback");
     }
     List<Procedure<Env>> subprocStack = procedureStack.getSubproceduresStack();
+    // Decide which procedures must keep the states they have executed before anything is rolled
+    // back: the decision reads the state history, which the rollback itself shrinks. Identity is
+    // used because procedures override equals with the data of the operation they run.
+    final Set<Procedure<Env>> proceduresKeepingTheirStates =
+        Collections.newSetFromMap(new IdentityHashMap<>());
+    for (Procedure<Env> procedure : subprocStack) {
+      // A null slot is a procedure whose record was deleted before the snapshot this stack was
+      // rebuilt from, so there is nothing left to roll back for it.
+      if (procedure != null && !procedure.isSuccess() && !procedure.isRollbackSupported()) {
+        proceduresKeepingTheirStates.add(procedure);
+      }
+    }
     int stackTail = subprocStack.size();
     while (stackTail-- > 0) {
       Procedure<Env> procedure = subprocStack.get(stackTail);
+      if (procedure == null) {
+        continue;
+      }
       if (procedure.isSuccess()) {
         subprocStack.remove(stackTail);
         cleanupAfterRollback(procedure);
         continue;
       }
+      if (proceduresKeepingTheirStates.contains(procedure)) {
+        // Finished below, once the rest of the stack has been rolled back.
+        continue;
+      }
       ProcedureLockState lockState = acquireLock(procedure);
       if (lockState != ProcedureLockState.LOCK_ACQUIRED) {
         return new LockStateResult<>(lockState, procedure);
@@ -730,12 +751,109 @@ private LockStateResult<Env> executeRootStackRollback(
       }
     }
 
+    if (!proceduresKeepingTheirStates.isEmpty()) {
+      // The children go first: releasing a child's lock writes its record back, which has to be
+      // covered by the root's cleanup. The root is finished last so that no write of a child
+      // follows the write which persists the root as finished.
+      for (Procedure<Env> procedure : proceduresKeepingTheirStates) {
+        if (procedure.hasParent()) {
+          finishWithoutRollback(procedure);
+        }
+      }
+      for (Procedure<Env> procedure : proceduresKeepingTheirStates) {
+        if (!procedure.hasParent()) {
+          finishWithoutRollback(procedure);
+        }
+      }
+      subprocStack.removeIf(proceduresKeepingTheirStates::contains);
+    }
+
     LOG.info(
         ProcedureMessages.ROLLED_BACK_TIME_DURATION_IS, rootProcedure, rootProcedure.elapsedTime());
     rootProcedureCleanup(rootProcedure);
     return new LockStateResult<>(ProcedureLockState.LOCK_ACQUIRED, rootProcedure);
   }
 
+  /**
+   * Finishes a procedure without undoing any of the states it has executed. The failure is still
+   * reported, as the operation the procedure was part of did not complete; what is dropped is only
+   * the bookkeeping of the rollback, so that neither this rollback nor the one a new leader
+   * recovers touches this procedure again.
+   */
+  private void finishWithoutRollback(final Procedure<Env> procedure) throws InterruptedException {
+    // The unlock persists the procedure through doReleaseLock(). It is done first, while the
+    // procedure is neither marked finished nor stripped of its state history, so that a snapshot
+    // taken right after this write is still a state recovery resumes. Forcing the release matters
+    // because the procedure is not finished yet, so releaseLock() would otherwise keep the lock of
+    // procedures which hold it across steps.
+    releaseLockIfHeld(procedure);
+    if (procedure.hasParent()) {
+      // Mark the procedure finished before dropping its state history, so that a worker which
+      // picks it up concurrently returns at the isFinished() check instead of rolling it back.
+      if (!procedure.isSuccess()) {
+        procedure.setState(ProcedureState.ROLLEDBACK);
+      }
+      procedure.abandonRollback();
+      try {
+        store.delete(procedure.getProcId());
+        // do not remove this procedure when exception occurred
+        procedures.remove(procedure.getProcId());
+      } catch (final Exception e) {
+        LOG.warn(ProcedureMessages.FAILED_TO_DELETE_PROCEDURE_ON_ROLLBACK, e);
+      }
+    } else {
+      // A child which never executed has no rollback step and is not visited above, so its record
+      // goes with the ones of the children which were.
+      // Every child record has to be gone before the root may be persisted as finished: a snapshot
+      // holding a finished root beside a child record makes recover() fail on that child. Deleting
+      // is idempotent, so it is retried until durable.
+      final long[] childProcIds = rollbackStack.get(procedure.getProcId()).getSubprocedureIds();
+      for (final long childProcId : childProcIds) {
+        RetryUtils.executeWithEndlessBackoffRetry(
+            () -> store.delete(childProcId),
+            ProcedureMessages.LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1);
+        // The retry helper also returns on interruption. Abort the entire rollback before
+        // marking the root finished or letting the caller clean it up.
+        if (Thread.currentThread().isInterrupted()) {
+          throw new InterruptedException();
+        }
+      }
+      if (!procedure.isSuccess()) {
+        procedure.setState(ProcedureState.ROLLEDBACK);
+      }
+      procedure.abandonRollback();
+      // The stack indexes and the states just dropped are persisted here, so that the rollback
+      // recovered by a new leader leaves this procedure out. Endless retry since this step is not
+      // idempotent.
+      RetryUtils.executeWithEndlessBackoffRetry(
+          () -> store.update(procedure),
+          ProcedureMessages.LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354);
+      if (Thread.currentThread().isInterrupted()) {
+        throw new InterruptedException();
+      }
+    }
+    procedure.updateMetricsOnFinish(getEnvironment(), procedure.elapsedTime(), false);
+    LOG.info(
+        ProcedureMessages
+            .LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9,
+        procedure);
+  }
+
+  /**
+   * The rollback loop skipped acquireLock() for procedures finished without rollback, so a lock
+   * restored by restoreLock() on load is still recorded in lockedWhenLoading instead of locked.
+   * Finish that bookkeeping and release the lock here: rootProcedureCleanup() releases only the
+   * locks that hasLock() reports, and a procedure finished without rollback must not keep holding
+   * its lock.
+   */
+  private void releaseLockIfHeld(final Procedure<Env> procedure) {
+    if (procedure.hasLock() || procedure.isLockedWhenLoading()) {
+      if (acquireLock(procedure) == ProcedureLockState.LOCK_ACQUIRED) {
+        releaseLock(procedure, true);
+      }
+    }
+  }
+
   private ProcedureLockState acquireLock(Procedure<Env> proc) {
     if (proc.hasLock()) {
       return ProcedureLockState.LOCK_ACQUIRED;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java

@@ -89,7 +89,9 @@ protected synchronized List<Procedure<Env>> getSubproceduresStack() {
   protected synchronized ProcedureException getException() {
     if (subprocStack != null) {
       for (Procedure<Env> proc : subprocStack) {
-        if (proc.hasException()) {
+        // Null slots come from procedures whose records were deleted before the stack was
+        // rebuilt, see ProcedureExecutor#executeRootStackRollback.
+        if (proc != null && proc.hasException()) {
           return proc.getException();
         }
       }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java

@@ -31,6 +31,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.ConcurrentLinkedDeque;
 
@@ -252,12 +253,49 @@ protected boolean isEofState() {
     return !states.isEmpty() && states.getLast() == EOF_STATE;
   }
 
+  @Override
+  protected synchronized void abandonRollback() {
+    super.abandonRollback();
+    states.clear();
+  }
+
   /**
-   * Used by the default implementation of abort() to know if the current state can be aborted and
-   * rollback can be triggered.
+   * Whether a failure in the specified state can be rolled back. A procedure asks this in its error
+   * handling to tell a fatal failure from a retriable one; the executor asks it once, through the
+   * overload without a state, to decide whether the procedure may be rolled back at all.
+   *
+   * <p>The executor asks with the state the procedure is at: the state which failed or is
+   * scheduled, or the last state executed when every state has already run. A procedure which
+   * returns false for the states after its commit point is therefore not rolled back once that
+   * point has been passed.
    */
   protected boolean isRollbackSupported(final TState state) {
-    return false;
+    return true;
+  }
+
+  /**
+   * Whether the states this procedure has executed may still be rolled back. The answer comes from
+   * the state the procedure is at, through the per-state overload above.
+   */
+  @Override
+  protected final boolean isRollbackSupported() {
+    return isRollbackSupported(getRollbackDecisionState());
+  }
+
+  /**
+   * The state a rollback decision is taken at. The last element of the history is the state which
+   * failed or is scheduled; once every state has run, the end marker is last and the decision falls
+   * on the state before it.
+   */
+  private TState getRollbackDecisionState() {
+    final Iterator<Integer> stateIds = states.descendingIterator();
+    while (stateIds.hasNext()) {
+      final int stateId = stateIds.next();
+      if (stateId != EOF_STATE) {
+        return getState(stateId);
+      }
+    }
+    return getInitialState();
   }
 
   private boolean noMoreState() {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveAINodeProcedure.java

@@ -135,6 +135,15 @@ protected void rollbackState(
     // no need to rollback
   }
 
+  /**
+   * A failure in either state is retried instead of failing the procedure, which has nothing to
+   * roll back.
+   */
+  @Override
+  protected boolean isRollbackSupported(final RemoveAINodeState state) {
+    return false;
+  }
+
   @Override
   protected RemoveAINodeState getState(int stateId) {
     return RemoveAINodeState.values()[stateId];

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java

@@ -162,6 +162,23 @@ private void setProperties(final ConfigNodeProcedureEnv env) {
   protected abstract ConfigPhysicalPlan createSetPropertiesPlan(
       Map<String, String> properties, boolean isRollback);
 
+  /**
+   * The properties are set on the ConfigNode once SET_PROPERTIES has completed, and the ConfigNode
+   * schema is the source of truth the DataNodes resync from. The procedure therefore stays
+   * rollbackable up to SET_PROPERTIES and not after it.
+   */
+  @Override
+  protected boolean isRollbackSupported(final SetTablePropertiesState state) {
+    switch (state) {
+      case VALIDATE_TABLE:
+      case PRE_RELEASE:
+      case SET_PROPERTIES:
+        return true;
+      default:
+        return false;
+    }
+  }
+
   @Override
   protected void rollbackState(
       final ConfigNodeProcedureEnv env, final SetTablePropertiesState state)

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java

@@ -154,6 +154,23 @@ protected String getActionMessage() {
     return "add table column";
   }
 
+  /**
+   * The column is part of the table on the ConfigNode once ADD_COLUMN has completed, and the
+   * ConfigNode schema is the source of truth the DataNodes resync from. The procedure therefore
+   * stays rollbackable up to ADD_COLUMN and not after it.
+   */
+  @Override
+  protected boolean isRollbackSupported(final AddTableColumnState state) {
+    switch (state) {
+      case COLUMN_CHECK:
+      case PRE_RELEASE:
+      case ADD_COLUMN:
+        return true;
+      default:
+        return false;
+    }
+  }
+
   @Override
   protected void rollbackState(final ConfigNodeProcedureEnv env, final AddTableColumnState state)
       throws IOException, InterruptedException, ProcedureException {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java

@@ -161,6 +161,7 @@ private void alterColumnDataType(final ConfigNodeProcedureEnv env) {
     setNextState(AlterTableColumnDataTypeState.COMMIT_RELEASE);
   }
 
+  /** The rollback of this procedure is a no-op, so it is never rolled back. */
   @Override
   protected boolean isRollbackSupported(final AlterTableColumnDataTypeState state) {
     return false;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java

@@ -212,7 +212,15 @@ private void commitReleaseTable(final ConfigNodeProcedureEnv env) {
 
   @Override
   protected boolean isRollbackSupported(final CreateTableState state) {
-    return true;
+    switch (state) {
+      case CHECK_TABLE_EXISTENCE:
+      case PRE_CREATE:
+      case PRE_RELEASE:
+      case COMMIT_CREATE:
+        return true;
+      default:
+        return false;
+    }
   }
 
   @Override

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java

@@ -233,6 +233,7 @@ private void dropColumn(final ConfigNodeProcedureEnv env) {
     }
   }
 
+  /** The rollback of this procedure is a no-op, so it is never rolled back. */
   @Override
   protected boolean isRollbackSupported(final DropTableColumnState state) {
     return false;