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,9 @@ 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";
 }

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

@@ -1515,4 +1515,8 @@ 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 = "结束流程（不回滚）";
 }

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;
@@ -700,6 +702,16 @@ 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) {
+      if (!procedure.isSuccess() && !procedure.isRollbackSupported()) {
+        proceduresKeepingTheirStates.add(procedure);
+      }
+    }
     int stackTail = subprocStack.size();
     while (stackTail-- > 0) {
       Procedure<Env> procedure = subprocStack.get(stackTail);
@@ -708,6 +720,10 @@ private LockStateResult<Env> executeRootStackRollback(
         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 +746,70 @@ private LockStateResult<Env> executeRootStackRollback(
       }
     }
 
+    if (!proceduresKeepingTheirStates.isEmpty()) {
+      subprocStack.removeIf(proceduresKeepingTheirStates::contains);
+      proceduresKeepingTheirStates.forEach(this::finishWithoutRollback);
+    }
+
     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) {
+    procedure.abandonRollback();
+    if (!procedure.isSuccess()) {
+      procedure.setState(ProcedureState.ROLLEDBACK);
+    }
+    // The rollback loop skipped acquireLock() for this procedure, 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 finished procedure must not keep holding its lock.
+    if (procedure.hasLock() || procedure.isLockedWhenLoading()) {
+      if (acquireLock(procedure) == ProcedureLockState.LOCK_ACQUIRED) {
+        releaseLock(procedure, false);
+      }
+    }
+    if (procedure.hasParent()) {
+      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.
+      final long[] childProcIds = rollbackStack.get(procedure.getProcId()).getSubprocedureIds();
+      if (childProcIds.length > 0) {
+        try {
+          store.delete(childProcIds);
+        } catch (final Exception e) {
+          LOG.warn(ProcedureMessages.FAILED_TO_DELETE_PROCEDURE_ON_ROLLBACK, e);
+        }
+      }
+      // 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);
+    }
+    procedure.updateMetricsOnFinish(getEnvironment(), procedure.elapsedTime(), false);
+    LOG.info(
+        ProcedureMessages
+            .LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9,
+        procedure);
+  }
+
   private ProcedureLockState acquireLock(Procedure<Env> proc) {
     if (proc.hasLock()) {
       return ProcedureLockState.LOCK_ACQUIRED;

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;

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

@@ -386,9 +386,22 @@ private void dropTable(final ConfigNodeProcedureEnv env) {
     }
   }
 
+  /**
+   * The data is deleted logically once DELETE_DATA has completed: the deletion is applied to every
+   * data region, and only the physical removal of the mod files continues in the background. The
+   * procedure is therefore not rolled back once it has moved on to DELETE_DEVICES. A view holds no
+   * data and goes straight to DROP_TABLE, see {@link DropViewProcedure}.
+   */
   @Override
   protected boolean isRollbackSupported(final DropTableState state) {
-    return state == DropTableState.CHECK_AND_INVALIDATE_TABLE || state == DropTableState.PRE_DELETE;
+    switch (state) {
+      case CHECK_AND_INVALIDATE_TABLE:
+      case PRE_DELETE:
+      case DELETE_DATA:
+        return true;
+      default:
+        return false;
+    }
   }
 
   @Override

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

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

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

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