Classify metadata migration restart state

2026-09-17 09:40:58 +00:00 · 2026-08-10 12:22:17 +08:00
parent 8efbb02603
commit 6f8258cc7d
5 changed files with 343 additions and 2 deletions
@@ -57,7 +57,6 @@ final class MigrationOperationTransitionPolicy {
            case ACTIVATING -> runningAt(next, MigrationStage.ACTIVATING, VerificationState.SUCCEEDED)
                    || rollingBackAt(next, MigrationRollbackOrigin.ACTIVATION_FAILURE)
                    || next.state() == MigrationOperationState.AWAITING_RESTART
-                    || next.state() == MigrationOperationState.SUCCEEDED
                    || failedWith(next, SetupErrorCode.MIGRATION_ACTIVATION_FAILED);
            case ROLLING_BACK -> rollbackContinues(current, next) || rollbackCompletes(current, next);
            default -> false;
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.workflow;
+
+import java.util.Objects;
+import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationStage;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ApplyMode;
+
+/**
+ * Classifies durable migration state after a restart without performing recovery I/O.
+ * Exact candidate evidence proves candidate identity, never the copy outcome.
+ */
+final class MigrationRestartClassifier {
+
+    Plan classify(MigrationOperationSnapshot snapshot, CandidateEvidence evidence) {
+        Objects.requireNonNull(snapshot, "snapshot");
+        Objects.requireNonNull(evidence, "evidence");
+        if (evidence == CandidateEvidence.INCONSISTENT
+                || evidence == CandidateEvidence.RECOVERY_REQUIRED) {
+            return Plan.RECOVERY_REQUIRED;
+        }
+        return snapshot.applyMode() == ApplyMode.MANAGED_WRITE
+                ? managed(snapshot, evidence) : external(snapshot, evidence);
+    }
+
+    private Plan managed(MigrationOperationSnapshot snapshot, CandidateEvidence evidence) {
+        if (evidence == CandidateEvidence.NOT_APPLICABLE) {
+            return Plan.RECOVERY_REQUIRED;
+        }
+        if (snapshot.terminal()) {
+            return evidence == CandidateEvidence.EXACT
+                    ? Plan.CLEANUP_TERMINAL_CANDIDATE : Plan.NONE;
+        }
+        return switch (snapshot.state()) {
+            case PENDING -> evidence == CandidateEvidence.EXACT
+                    ? Plan.RESUME_PREPARATION : Plan.CREDENTIALS_REQUIRED_FOR_PREPARATION;
+            case RUNNING -> managedRunning(snapshot.stage(), evidence);
+            case READY_TO_ACTIVATE -> exact(evidence, Plan.HOLD_READY_UNDER_STARTUP_GATE);
+            case AWAITING_RESTART -> exact(evidence, Plan.VERIFY_RESTART_CONVERGENCE);
+            case AWAITING_EXTERNAL_APPLY -> Plan.RECOVERY_REQUIRED;
+            case SUCCEEDED, FAILED, ROLLED_BACK -> Plan.RECOVERY_REQUIRED;
+        };
+    }
+
+    private Plan managedRunning(MigrationStage stage, CandidateEvidence evidence) {
+        if (evidence != CandidateEvidence.EXACT) {
+            return Plan.RECOVERY_REQUIRED;
+        }
+        return switch (stage) {
+            case COPYING, VERIFYING -> Plan.VERIFY_COPY_OUTCOME;
+            case ACTIVATING -> Plan.RECOVER_ACTIVATION;
+            case ROLLING_BACK -> Plan.RECOVER_ROLLBACK;
+            default -> Plan.RECOVERY_REQUIRED;
+        };
+    }
+
+    private Plan external(MigrationOperationSnapshot snapshot, CandidateEvidence evidence) {
+        if (evidence != CandidateEvidence.NOT_APPLICABLE) {
+            return Plan.RECOVERY_REQUIRED;
+        }
+        if (snapshot.terminal()) {
+            return Plan.NONE;
+        }
+        return switch (snapshot.state()) {
+            case PENDING -> Plan.CREDENTIALS_REQUIRED_FOR_PREPARATION;
+            case RUNNING -> snapshot.stage() == MigrationStage.COPYING
+                    || snapshot.stage() == MigrationStage.VERIFYING
+                    ? Plan.CREDENTIALS_REQUIRED_FOR_COPY_VERIFICATION : Plan.RECOVERY_REQUIRED;
+            case AWAITING_EXTERNAL_APPLY, AWAITING_RESTART -> Plan.VERIFY_RESTART_CONVERGENCE;
+            case READY_TO_ACTIVATE -> Plan.RECOVERY_REQUIRED;
+            case SUCCEEDED, FAILED, ROLLED_BACK -> Plan.RECOVERY_REQUIRED;
+        };
+    }
+
+    private Plan exact(CandidateEvidence evidence, Plan plan) {
+        return evidence == CandidateEvidence.EXACT ? plan : Plan.RECOVERY_REQUIRED;
+    }
+
+    enum CandidateEvidence {
+        NOT_APPLICABLE,
+        MISSING,
+        EXACT,
+        INCONSISTENT,
+        RECOVERY_REQUIRED
+    }
+
+    enum Plan {
+        NONE,
+        CLEANUP_TERMINAL_CANDIDATE,
+        RESUME_PREPARATION,
+        CREDENTIALS_REQUIRED_FOR_PREPARATION,
+        CREDENTIALS_REQUIRED_FOR_COPY_VERIFICATION,
+        VERIFY_COPY_OUTCOME,
+        HOLD_READY_UNDER_STARTUP_GATE,
+        RECOVER_ACTIVATION,
+        VERIFY_RESTART_CONVERGENCE,
+        RECOVER_ROLLBACK,
+        RECOVERY_REQUIRED
+    }
+}
@@ -406,7 +406,14 @@ class FileMigrationOperationStoreTest {
                VerificationState.SUCCEEDED, null, null, 1000, false, false, false,
                pending.targetIdentityHash(), pending.managedCandidateGeneration());
        store.compareAndTransition(pending.operationId(), MigrationOperationState.READY_TO_ACTIVATE, activating);
-        store.compareAndTransition(pending.operationId(), MigrationOperationState.RUNNING, succeeded(pending));
+        MigrationOperationSnapshot awaitingRestart = new MigrationOperationSnapshot(
+                pending.operationId(), MigrationOperationState.AWAITING_RESTART, pending.target(), pending.applyMode(),
+                MigrationStage.AWAITING_RESTART, 100, pending.createdAt(), pending.createdAt().plusSeconds(1), null,
+                VerificationState.SUCCEEDED, null, null, 1000, false, true, false,
+                pending.targetIdentityHash(), pending.managedCandidateGeneration());
+        store.compareAndTransition(pending.operationId(), MigrationOperationState.RUNNING, awaitingRestart);
+        store.compareAndTransition(
+                pending.operationId(), MigrationOperationState.AWAITING_RESTART, succeeded(pending));
    }

    private static void assertStoreError(SetupErrorCode expected, ThrowingAction action) {
@@ -56,6 +56,21 @@ class MigrationOperationTransitionPolicyTest {
        assertAllowed(restart, succeeded);
    }

+    @Test
+    void managedActivationMustPersistRestartBeforeSuccess() {
+        MigrationOperationSnapshot activating = snapshot(MigrationOperationState.RUNNING, MigrationStage.ACTIVATING,
+                100, STARTED, null, VerificationState.SUCCEEDED, null, 1000, false, false, false);
+        MigrationOperationSnapshot restart = snapshot(MigrationOperationState.AWAITING_RESTART,
+                MigrationStage.AWAITING_RESTART, 100, STARTED, null, VerificationState.SUCCEEDED,
+                null, 1000, false, true, false);
+        MigrationOperationSnapshot succeeded = snapshot(MigrationOperationState.SUCCEEDED, MigrationStage.COMPLETED,
+                100, STARTED, COMPLETED, VerificationState.SUCCEEDED, null, 0, false, false, false);
+
+        assertRejected(activating, succeeded);
+        assertAllowed(activating, restart);
+        assertAllowed(restart, succeeded);
+    }
+
    @Test
    void acceptsExternalFailedAndRolledBackExitsButTerminalStatesStayClosed() {
        MigrationOperationSnapshot verifying = external(MigrationOperationState.RUNNING, MigrationStage.VERIFYING,
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.workflow;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.CandidateEvidence.EXACT;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.CLEANUP_TERMINAL_CANDIDATE;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.CREDENTIALS_REQUIRED_FOR_COPY_VERIFICATION;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.CREDENTIALS_REQUIRED_FOR_PREPARATION;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.HOLD_READY_UNDER_STARTUP_GATE;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.NONE;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.RECOVER_ACTIVATION;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.RECOVER_ROLLBACK;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.RESUME_PREPARATION;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.VERIFY_COPY_OUTCOME;
+import static org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan.VERIFY_RESTART_CONVERGENCE;
+
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.Set;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationOperationState;
+import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationStage;
+import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationTarget;
+import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.VerificationState;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ApplyMode;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.SetupErrorCode;
+import org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.CandidateEvidence;
+import org.apache.hertzbeat.manager.setup.workflow.MigrationRestartClassifier.Plan;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class MigrationRestartClassifierTest {
+
+    private static final String TARGET_IDENTITY_HASH =
+            "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
+    private static final String CANDIDATE_GENERATION = "candidate-generation-1";
+    private static final Instant CREATED = Instant.parse("2026-08-10T01:00:00Z");
+    private static final Instant STARTED = CREATED.plusSeconds(1);
+    private static final Instant COMPLETED = STARTED.plusSeconds(1);
+    private static final Plan R = Plan.RECOVERY_REQUIRED;
+    private final MigrationRestartClassifier classifier = new MigrationRestartClassifier();
+
+    @ParameterizedTest(name = "{0}-{1}-{2}-{3}")
+    @MethodSource("restartCases")
+    void classifiesEveryDurablePhaseAndEvidence(
+            ApplyMode mode, MigrationOperationState state, MigrationStage stage,
+            CandidateEvidence evidence, Plan expected) {
+        assertThat(classifier.classify(snapshot(mode, state, stage, progress(stage)), evidence))
+                .isEqualTo(expected);
+    }
+
+    @Test
+    void copyProgressDoesNotChangeRecoveryPlan() {
+        for (int progress : new int[] {0, 10, 99}) {
+            MigrationOperationSnapshot snapshot = snapshot(
+                    ApplyMode.MANAGED_WRITE, MigrationOperationState.RUNNING, MigrationStage.COPYING, progress);
+            assertThat(classifier.classify(snapshot, EXACT)).isEqualTo(VERIFY_COPY_OUTCOME);
+        }
+    }
+
+    @Test
+    void classificationNamesExposeNoConfigurationOrDataIdentity() {
+        Set<String> forbidden = Set.of("jdbc", "url", "user", "password", "table", "checksum");
+        assertThat(Arrays.stream(Plan.values()).map(Enum::name).map(value -> value.toLowerCase(Locale.ROOT)))
+                .allSatisfy(value -> assertThat(forbidden).noneMatch(value::contains));
+        assertThat(Arrays.stream(CandidateEvidence.values()).map(Enum::name)
+                .map(value -> value.toLowerCase(Locale.ROOT)))
+                .allSatisfy(value -> assertThat(forbidden).noneMatch(value::contains));
+    }
+
+    @Test
+    void credentialPlansCannotTurnVerificationRecoveryIntoCopyExecution() {
+        assertThat(CREDENTIALS_REQUIRED_FOR_COPY_VERIFICATION)
+                .isNotEqualTo(CREDENTIALS_REQUIRED_FOR_PREPARATION);
+        assertThat(CREDENTIALS_REQUIRED_FOR_COPY_VERIFICATION.name())
+                .contains("VERIFICATION")
+                .doesNotContain("RESUME", "COPY_EXECUTION");
+    }
+
+    private static Stream<Arguments> restartCases() {
+        return Stream.concat(managedCases(), externalCases());
+    }
+
+    private static Stream<Arguments> managedCases() {
+        return Stream.of(
+                phase(ApplyMode.MANAGED_WRITE, MigrationOperationState.PENDING, MigrationStage.QUEUED,
+                        R, CREDENTIALS_REQUIRED_FOR_PREPARATION, RESUME_PREPARATION, R, R),
+                phase(ApplyMode.MANAGED_WRITE, MigrationOperationState.RUNNING, MigrationStage.COPYING,
+                        R, R, VERIFY_COPY_OUTCOME, R, R),
+                phase(ApplyMode.MANAGED_WRITE, MigrationOperationState.RUNNING, MigrationStage.VERIFYING,
+                        R, R, VERIFY_COPY_OUTCOME, R, R),
+                phase(ApplyMode.MANAGED_WRITE, MigrationOperationState.READY_TO_ACTIVATE,
+                        MigrationStage.READY_TO_ACTIVATE, R, R, HOLD_READY_UNDER_STARTUP_GATE, R, R),
+                phase(ApplyMode.MANAGED_WRITE, MigrationOperationState.RUNNING, MigrationStage.ACTIVATING,
+                        R, R, RECOVER_ACTIVATION, R, R),
+                phase(ApplyMode.MANAGED_WRITE, MigrationOperationState.AWAITING_RESTART,
+                        MigrationStage.AWAITING_RESTART, R, R, VERIFY_RESTART_CONVERGENCE, R, R),
+                phase(ApplyMode.MANAGED_WRITE, MigrationOperationState.RUNNING, MigrationStage.ROLLING_BACK,
+                        R, R, RECOVER_ROLLBACK, R, R),
+                phase(ApplyMode.MANAGED_WRITE, MigrationOperationState.AWAITING_EXTERNAL_APPLY,
+                        MigrationStage.AWAITING_EXTERNAL_APPLY, R, R, R, R, R),
+                terminal(ApplyMode.MANAGED_WRITE, MigrationOperationState.SUCCEEDED, MigrationStage.COMPLETED,
+                        R, NONE, CLEANUP_TERMINAL_CANDIDATE, R, R),
+                terminal(ApplyMode.MANAGED_WRITE, MigrationOperationState.FAILED, MigrationStage.FAILED,
+                        R, NONE, CLEANUP_TERMINAL_CANDIDATE, R, R),
+                terminal(ApplyMode.MANAGED_WRITE, MigrationOperationState.ROLLED_BACK,
+                        MigrationStage.ROLLED_BACK, R, NONE, CLEANUP_TERMINAL_CANDIDATE, R, R))
+                .flatMap(stream -> stream);
+    }
+
+    private static Stream<Arguments> externalCases() {
+        return Stream.of(
+                phase(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.PENDING, MigrationStage.QUEUED,
+                        CREDENTIALS_REQUIRED_FOR_PREPARATION, R, R, R, R),
+                phase(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.RUNNING, MigrationStage.COPYING,
+                        CREDENTIALS_REQUIRED_FOR_COPY_VERIFICATION, R, R, R, R),
+                phase(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.RUNNING, MigrationStage.VERIFYING,
+                        CREDENTIALS_REQUIRED_FOR_COPY_VERIFICATION, R, R, R, R),
+                phase(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.AWAITING_EXTERNAL_APPLY,
+                        MigrationStage.AWAITING_EXTERNAL_APPLY, VERIFY_RESTART_CONVERGENCE, R, R, R, R),
+                phase(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.AWAITING_RESTART,
+                        MigrationStage.AWAITING_RESTART, VERIFY_RESTART_CONVERGENCE, R, R, R, R),
+                phase(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.READY_TO_ACTIVATE,
+                        MigrationStage.READY_TO_ACTIVATE, R, R, R, R, R),
+                phase(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.RUNNING, MigrationStage.ACTIVATING,
+                        R, R, R, R, R),
+                phase(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.RUNNING, MigrationStage.ROLLING_BACK,
+                        R, R, R, R, R),
+                terminal(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.SUCCEEDED, MigrationStage.COMPLETED,
+                        NONE, R, R, R, R),
+                terminal(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.FAILED, MigrationStage.FAILED,
+                        NONE, R, R, R, R),
+                terminal(ApplyMode.EXTERNAL_APPLY, MigrationOperationState.ROLLED_BACK,
+                        MigrationStage.ROLLED_BACK, NONE, R, R, R, R))
+                .flatMap(stream -> stream);
+    }
+
+    private static Stream<Arguments> phase(
+            ApplyMode mode, MigrationOperationState state, MigrationStage stage, Plan... expected) {
+        CandidateEvidence[] evidence = CandidateEvidence.values();
+        return IntStream.range(0, evidence.length)
+                .mapToObj(index -> Arguments.of(mode, state, stage, evidence[index], expected[index]));
+    }
+
+    private static Stream<Arguments> terminal(
+            ApplyMode mode, MigrationOperationState state, MigrationStage stage, Plan... expected) {
+        return phase(mode, state, stage, expected);
+    }
+
+    private static MigrationOperationSnapshot snapshot(
+            ApplyMode mode, MigrationOperationState state, MigrationStage stage, int progress) {
+        VerificationState verification = verification(stage);
+        SetupErrorCode error = error(state);
+        MigrationRollbackOrigin rollback = rollback(state, stage);
+        return new MigrationOperationSnapshot(
+                "migration-1", state, MigrationTarget.MYSQL, mode, stage, progress, CREATED,
+                state == MigrationOperationState.PENDING ? null : STARTED,
+                terminal(state) ? COMPLETED : null, verification, error, rollback,
+                polling(state), state == MigrationOperationState.READY_TO_ACTIVATE,
+                state == MigrationOperationState.AWAITING_RESTART,
+                state == MigrationOperationState.AWAITING_EXTERNAL_APPLY,
+                TARGET_IDENTITY_HASH, mode == ApplyMode.MANAGED_WRITE ? CANDIDATE_GENERATION : null);
+    }
+
+    private static int progress(MigrationStage stage) {
+        return switch (stage) {
+            case QUEUED -> 0;
+            case COPYING, FAILED -> 10;
+            default -> 100;
+        };
+    }
+
+    private static VerificationState verification(MigrationStage stage) {
+        return switch (stage) {
+            case QUEUED, COPYING, FAILED -> VerificationState.PENDING;
+            case VERIFYING -> VerificationState.RUNNING;
+            default -> VerificationState.SUCCEEDED;
+        };
+    }
+
+    private static SetupErrorCode error(MigrationOperationState state) {
+        return switch (state) {
+            case FAILED -> SetupErrorCode.MIGRATION_COPY_FAILED;
+            case ROLLED_BACK -> SetupErrorCode.MIGRATION_ACTIVATION_FAILED;
+            default -> null;
+        };
+    }
+
+    private static MigrationRollbackOrigin rollback(
+            MigrationOperationState state, MigrationStage stage) {
+        return state == MigrationOperationState.ROLLED_BACK || stage == MigrationStage.ROLLING_BACK
+                ? MigrationRollbackOrigin.ACTIVATION_FAILURE : null;
+    }
+
+    private static long polling(MigrationOperationState state) {
+        return state == MigrationOperationState.PENDING || state == MigrationOperationState.RUNNING
+                || state == MigrationOperationState.AWAITING_RESTART ? 1000 : 0;
+    }
+
+    private static boolean terminal(MigrationOperationState state) {
+        return state == MigrationOperationState.SUCCEEDED
+                || state == MigrationOperationState.FAILED
+                || state == MigrationOperationState.ROLLED_BACK;
+    }
+}