feat(core): add stage-level runAlways support

build.gradle.kts

@@ -24,7 +24,7 @@ allprojects {
 
     extra["generalUtilVersion"] = "1.6.0"
     extra["templateApiVersion"] = "1.3.4"
-    extra["coreApiVersion"] = "1.3.3"
+    extra["coreApiVersion"] = "1.4.0-SNAPSHOT"
     extra["sqlVersion"] = "1.3.2"
     extra["mongodbTemplateVersion"] = "1.3.2"
 

cloud/flamingock-cloud-api/build.gradle.kts

@@ -7,6 +7,7 @@ val coreApiVersion: String by extra
 val jacksonVersion = "2.14.1"
 dependencies {
     api("io.flamingock:flamingock-core-api:${coreApiVersion}")
+    implementation("com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}")
     testImplementation("com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}")
 }
 

cloud/flamingock-cloud-api/src/main/java/io/flamingock/cloud/api/request/ChangeRequest.java

@@ -44,35 +44,33 @@ public class ChangeRequest {
 
     private boolean transactional;
 
+    private boolean runAlways;
+
     public ChangeRequest() {
     }
 
     public static ChangeRequest change(String id, boolean transactional) {
-        return new ChangeRequest(id, CloudTargetSystemAuditMarkType.NONE, null, transactional);
+        return new ChangeRequest(id, CloudTargetSystemAuditMarkType.NONE, null, transactional, false);
     }
 
     public static ChangeRequest ongoingExecution(String id, boolean transactional) {
-        return new ChangeRequest(id, CloudTargetSystemAuditMarkType.APPLIED, null, transactional);
+        return new ChangeRequest(id, CloudTargetSystemAuditMarkType.APPLIED, null, transactional, false);
     }
 
     public static ChangeRequest ongoingRollback(String id, boolean transactional) {
-        return new ChangeRequest(id, CloudTargetSystemAuditMarkType.ROLLED_BACK, null, transactional);
-    }
-
-    public ChangeRequest(String id,
-                         CloudTargetSystemAuditMarkType ongoingStatus,
-                         boolean transactional) {
-        this(id, ongoingStatus, null, transactional);
+        return new ChangeRequest(id, CloudTargetSystemAuditMarkType.ROLLED_BACK, null, transactional, false);
     }
 
     public ChangeRequest(String id,
                          CloudTargetSystemAuditMarkType ongoingStatus,
                          CloudChangeStatus currentStatus,
-                         boolean transactional) {
+                         boolean transactional,
+                         boolean runAlways) {
         this.id = id;
         this.ongoingStatus = ongoingStatus;
         this.currentStatus = currentStatus;
         this.transactional = transactional;
+        this.runAlways = runAlways;
     }
 
     public String getId() {
@@ -91,6 +89,10 @@ public boolean isTransactional() {
         return transactional;
     }
 
+    public boolean isRunAlways() {
+        return runAlways;
+    }
+
     public void setId(String id) {
         this.id = id;
     }
@@ -107,19 +109,24 @@ public void setTransactional(boolean transactional) {
         this.transactional = transactional;
     }
 
+    public void setRunAlways(boolean runAlways) {
+        this.runAlways = runAlways;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         ChangeRequest that = (ChangeRequest) o;
         return transactional == that.transactional
+                && runAlways == that.runAlways
                 && java.util.Objects.equals(id, that.id)
                 && java.util.Objects.equals(ongoingStatus, that.ongoingStatus)
                 && java.util.Objects.equals(currentStatus, that.currentStatus);
     }
 
     @Override
     public int hashCode() {
-        return java.util.Objects.hash(id, ongoingStatus, currentStatus, transactional);
+        return java.util.Objects.hash(id, ongoingStatus, currentStatus, transactional, runAlways);
     }
-}
+}

cloud/flamingock-cloud-api/src/test/java/io/flamingock/cloud/api/request/ClientSubmissionRequestSerializationTest.java

@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.flamingock.api.StageType;
+import io.flamingock.cloud.api.vo.CloudChangeStatus;
 import io.flamingock.cloud.api.vo.CloudStageStatus;
 import io.flamingock.cloud.api.vo.CloudTargetSystemAuditMarkType;
 import org.junit.jupiter.api.DisplayName;
@@ -49,12 +50,10 @@ class ClientSubmissionRequestSerializationTest {
     void serializesBlocksWithTypeAndStages() throws Exception {
         StageRequest sysStage = new StageRequest(
                 "system-stage", 0, CloudStageStatus.NOT_STARTED,
-                Collections.singletonList(new ChangeRequest("sys-c1",
-                        CloudTargetSystemAuditMarkType.NONE, false)));
+                Collections.singletonList(new ChangeRequest("sys-c1", CloudTargetSystemAuditMarkType.NONE, null, false, false)));
         StageRequest userStage = new StageRequest(
                 "user-stage", 1, CloudStageStatus.NOT_STARTED,
-                Collections.singletonList(new ChangeRequest("user-c1",
-                        CloudTargetSystemAuditMarkType.NONE, false)));
+                Collections.singletonList(new ChangeRequest("user-c1", CloudTargetSystemAuditMarkType.NONE, null, false, false)));
 
         ClientSubmissionRequest request = new ClientSubmissionRequest(Arrays.asList(
                 new StageBlockRequest(StageType.SYSTEM, Collections.singletonList(sysStage)),
@@ -82,15 +81,12 @@ void roundTripsPreservingBlockOrderAndContents() throws Exception {
         ClientSubmissionRequest original = new ClientSubmissionRequest(Arrays.asList(
                 new StageBlockRequest(StageType.LEGACY, Collections.singletonList(
                         new StageRequest("legacy", 0, CloudStageStatus.COMPLETED,
-                                Collections.singletonList(new ChangeRequest("legacy-c1",
-                                        CloudTargetSystemAuditMarkType.APPLIED, true))))),
+                                Collections.singletonList(new ChangeRequest("legacy-c1", CloudTargetSystemAuditMarkType.APPLIED, null, true, false))))),
                 new StageBlockRequest(StageType.DEFAULT, Arrays.asList(
                         new StageRequest("user-a", 1, CloudStageStatus.STARTED,
-                                Collections.singletonList(new ChangeRequest("user-a-c1",
-                                        CloudTargetSystemAuditMarkType.NONE, false))),
+                                Collections.singletonList(new ChangeRequest("user-a-c1", CloudTargetSystemAuditMarkType.NONE, null, false, false))),
                         new StageRequest("user-b", 2, CloudStageStatus.NOT_STARTED,
-                                Collections.singletonList(new ChangeRequest("user-b-c1",
-                                        CloudTargetSystemAuditMarkType.NONE, false)))))));
+                                Collections.singletonList(new ChangeRequest("user-b-c1", CloudTargetSystemAuditMarkType.NONE, null, false, false)))))));
 
         String json = MAPPER.writeValueAsString(original);
         ClientSubmissionRequest deserialized = MAPPER.readValue(json, ClientSubmissionRequest.class);
@@ -109,11 +105,9 @@ void roundTripsPreservingBlockOrderAndContents() throws Exception {
     @DisplayName("Same StageType repeated across multiple blocks is preserved on the wire (multi-block-same-type lock-in)")
     void sameStageTypeAcrossMultipleBlocksIsPreserved() throws Exception {
         StageRequest a = new StageRequest("user-a", 0, CloudStageStatus.NOT_STARTED,
-                Collections.singletonList(new ChangeRequest("a-c1",
-                        CloudTargetSystemAuditMarkType.NONE, false)));
+                Collections.singletonList(new ChangeRequest("a-c1", CloudTargetSystemAuditMarkType.NONE, null, false, false)));
         StageRequest b = new StageRequest("user-b", 1, CloudStageStatus.NOT_STARTED,
-                Collections.singletonList(new ChangeRequest("b-c1",
-                        CloudTargetSystemAuditMarkType.NONE, false)));
+                Collections.singletonList(new ChangeRequest("b-c1", CloudTargetSystemAuditMarkType.NONE, null, false, false)));
 
         ClientSubmissionRequest request = new ClientSubmissionRequest(Arrays.asList(
                 new StageBlockRequest(StageType.DEFAULT, Collections.singletonList(a)),
@@ -142,6 +136,23 @@ void emptyBlocksListRoundTrips() throws Exception {
         assertTrue(deserialized.getBlocks().isEmpty());
     }
 
+    @Test
+    @DisplayName("ChangeRequest round-trips runAlways flag")
+    void changeRequestRoundTripsRunAlwaysFlag() throws Exception {
+        ChangeRequest change = new ChangeRequest(
+                "repeatable-c1",
+                CloudTargetSystemAuditMarkType.APPLIED,
+                CloudChangeStatus.ALREADY_APPLIED,
+                false,
+                true);
+
+        String json = MAPPER.writeValueAsString(change);
+        ChangeRequest deserialized = MAPPER.readValue(json, ChangeRequest.class);
+
+        assertTrue(deserialized.isRunAlways());
+        assertEquals(change, deserialized);
+    }
+
     @Test
     @DisplayName("Old flat 'stages' field at top level is silently ignored (clean cut — no fallback)")
     void oldFlatStagesFieldIsIgnored() throws Exception {

cloud/flamingock-cloud/src/main/java/io/flamingock/cloud/planner/CloudExecutionPlanMapper.java

@@ -115,7 +115,7 @@ private static ChangeRequest mapToChangeRequest(AbstractLoadedChange descriptor,
                 : CloudTargetSystemAuditMarkType.NONE;
         CloudChangeStatus currentStatus = CloudApiMapper.toCloud(
                 currentStatusByChangeId.get(descriptor.getId()));
-        return new ChangeRequest(descriptor.getId(), cloudStatus, currentStatus, descriptor.isTransactional());
+        return new ChangeRequest(descriptor.getId(), cloudStatus, currentStatus, descriptor.isTransactional(), descriptor.isRunAlways());
     }
 
     static List<ExecutableStage> getExecutableStages(ExecutionPlanResponse response, List<AbstractLoadedStage> loadedStages) {
@@ -153,8 +153,10 @@ private static ChangeActionMap getChangeActionMap(AbstractLoadedStage loadedStag
             String changeId = change.getId();
             CloudChangeAction cloudAction = actionsMapByChangeId.get(changeId);
 
-            // If change not in response, assume it's already applied (cloud orchestrator decision)
-            if (cloudAction == null) {
+            if (change.isRunAlways() && (cloudAction == null || cloudAction == CloudChangeAction.SKIP)) {
+                actionMap.put(changeId, ChangeAction.APPLY);
+            } else if (cloudAction == null) {
+                // If change not in response, assume it's already applied (cloud orchestrator decision)
                 actionMap.put(changeId, ChangeAction.SKIP);
             } else {
                 // Use anti-corruption layer to map cloud domain to internal domain

cloud/flamingock-cloud/src/test/java/io/flamingock/cloud/planner/CloudExecutionPlanMapperTest.java

@@ -16,6 +16,7 @@
 package io.flamingock.cloud.planner;
 
 import io.flamingock.api.StageType;
+import io.flamingock.api.StageMode;
 import io.flamingock.cloud.api.response.ChangeResponse;
 import io.flamingock.cloud.api.response.ExecutionPlanResponse;
 import io.flamingock.cloud.api.response.StageResponse;
@@ -57,6 +58,8 @@
 import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 class CloudExecutionPlanMapperTest {
 
@@ -130,6 +133,31 @@ void shouldDefaultToSkipWhenChangeNotInResponse() {
         assertEquals(ChangeAction.SKIP, actions.get(change2.getId()));
     }
 
+    @Test
+    @DisplayName("Should apply runAlways change when cloud response omits it")
+    void shouldApplyRunAlwaysChangeWhenCloudResponseOmitsIt() {
+        AbstractLoadedChange runAlwaysChange = mockChange("run-always", true);
+        CapturingStage stage = new CapturingStage("stage-1", runAlwaysChange);
+        ExecutionPlanResponse response = buildResponse(buildStageResponse("stage-1", 0));
+
+        List<ExecutableStage> result = CloudExecutionPlanMapper.getExecutableStages(response, Collections.singletonList(stage));
+
+        assertEquals(ChangeAction.APPLY, result.get(0).getChanges().get(0).getAction());
+    }
+
+    @Test
+    @DisplayName("Should apply runAlways change when cloud response asks to skip it")
+    void shouldApplyRunAlwaysChangeWhenCloudResponseSkipsIt() {
+        AbstractLoadedChange runAlwaysChange = mockChange("run-always", true);
+        CapturingStage stage = new CapturingStage("stage-1", runAlwaysChange);
+        ExecutionPlanResponse response = buildResponse(
+                buildStageResponse("stage-1", 0, changeResponse("run-always", CloudChangeAction.SKIP)));
+
+        List<ExecutableStage> result = CloudExecutionPlanMapper.getExecutableStages(response, Collections.singletonList(stage));
+
+        assertEquals(ChangeAction.APPLY, result.get(0).getChanges().get(0).getAction());
+    }
+
     @Test
     @DisplayName("Should correctly map mixed actions in a single stage")
     void shouldMapMixedActions() {
@@ -220,6 +248,19 @@ void shouldMapOngoingStatusFromAuditMarksToChangeRequests() {
         assertEquals(CloudTargetSystemAuditMarkType.NONE, marksByChangeId.get(change2.getId()));
     }
 
+    @Test
+    @DisplayName("toRequest() maps runAlways from loaded changes to ChangeRequest")
+    void shouldMapRunAlwaysToChangeRequest() {
+        AbstractLoadedChange runAlwaysChange = mockChange("run-always", true);
+        AbstractLoadedStage stage = new CapturingStage("stage-1", runAlwaysChange);
+
+        ExecutionPlanRequest request = CloudExecutionPlanMapper.toRequest(
+                PipelineRun.of(Collections.singletonList(stage)), 60000L, Collections.emptyMap());
+
+        ChangeRequest changeRequest = request.getClientSubmission().getBlocks().get(0).getStages().get(0).getChanges().get(0);
+        assertTrue(changeRequest.isRunAlways());
+    }
+
     @Test
     @DisplayName("toRequest() populates ChangeRequest.currentStatus from the operation's recorded ChangeResult statuses")
     void shouldMapCurrentStatusFromPipelineRun() {
@@ -303,12 +344,9 @@ void shouldMapPerStageStatusFromPipelineRun() {
     void toRequestEmitsOneBlockPerStageRunBlock() {
         // Three stages of three different types — PipelineRun.of(...) groups them into
         // SYSTEM -> LEGACY -> DEFAULT blocks (one stage each).
-        AbstractLoadedStage systemStage = new DefaultLoadedStage("system-stage", StageType.SYSTEM,
-                Collections.singletonList(change1));
-        AbstractLoadedStage legacyStage = new DefaultLoadedStage("legacy-stage", StageType.LEGACY,
-                Collections.singletonList(change1));
-        AbstractLoadedStage userStage = new DefaultLoadedStage("user-stage", StageType.DEFAULT,
-                Collections.singletonList(change1));
+        AbstractLoadedStage systemStage = new io.flamingock.internal.core.pipeline.loaded.stage.SystemLoadedStage("system-stage", StageType.SYSTEM, Collections.singletonList(change1));
+        AbstractLoadedStage legacyStage = new io.flamingock.internal.core.pipeline.loaded.stage.LegacyLoadedStage("legacy-stage", StageType.LEGACY, Collections.singletonList(change1));
+        AbstractLoadedStage userStage = new DefaultLoadedStage("user-stage", StageType.DEFAULT, StageMode.STANDARD, Collections.singletonList(change1));
 
         PipelineRun pipelineRun = PipelineRun.of(Arrays.asList(userStage, systemStage, legacyStage));
 
@@ -333,12 +371,9 @@ void toRequestEmitsOneBlockPerStageRunBlock() {
     void toRequestPreservesGlobalStageOrderAcrossBlocks() {
         // Two stages in the SYSTEM block, then one in DEFAULT — verify the per-stage `order`
         // field increments globally across the flattened block sequence.
-        AbstractLoadedStage system1 = new DefaultLoadedStage("system-1", StageType.SYSTEM,
-                Collections.singletonList(change1));
-        AbstractLoadedStage system2 = new DefaultLoadedStage("system-2", StageType.SYSTEM,
-                Collections.singletonList(change2));
-        AbstractLoadedStage user1 = new DefaultLoadedStage("user-1", StageType.DEFAULT,
-                Collections.singletonList(change1));
+        AbstractLoadedStage system1 = new io.flamingock.internal.core.pipeline.loaded.stage.SystemLoadedStage("system-1", StageType.SYSTEM, Collections.singletonList(change1));
+        AbstractLoadedStage system2 = new io.flamingock.internal.core.pipeline.loaded.stage.SystemLoadedStage("system-2", StageType.SYSTEM, Collections.singletonList(change2));
+        AbstractLoadedStage user1 = new DefaultLoadedStage("user-1", StageType.DEFAULT, StageMode.STANDARD, Collections.singletonList(change1));
 
         PipelineRun pipelineRun = PipelineRun.of(Arrays.asList(system1, system2, user1));
 
@@ -368,7 +403,39 @@ void toRequestEmptyPipelineRunYieldsEmptyBlocks() {
     }
 
     private static DefaultLoadedStage buildStage(String name, AbstractLoadedChange... changes) {
-        return new DefaultLoadedStage(name, StageType.DEFAULT, Arrays.asList(changes));
+        return new DefaultLoadedStage(name, StageType.DEFAULT, StageMode.STANDARD, Arrays.asList(changes));
+    }
+
+    private static AbstractLoadedChange mockChange(String id, boolean runAlways) {
+        AbstractLoadedChange change = mock(AbstractLoadedChange.class);
+        when(change.getId()).thenReturn(id);
+        when(change.isRunAlways()).thenReturn(runAlways);
+        when(change.isTransactional()).thenReturn(false);
+        when(change.isStandard()).thenReturn(true);
+        return change;
+    }
+
+    private static class CapturingStage extends DefaultLoadedStage {
+
+        CapturingStage(String name, AbstractLoadedChange... changes) {
+            super(name, StageType.DEFAULT, StageMode.STANDARD, Arrays.asList(changes));
+        }
+
+        @Override
+        public ExecutableStage applyActions(io.flamingock.internal.common.core.recovery.action.ChangeActionMap actionPlan) {
+            List<ExecutableChange> executableChanges = getChanges().stream()
+                    .map(change -> executableChange(change.getId(), actionPlan.getActionFor(change.getId())))
+                    .collect(Collectors.toList());
+            return new ExecutableStage(getName(), executableChanges);
+        }
+    }
+
+    private static ExecutableChange executableChange(String id, ChangeAction action) {
+        ExecutableChange executableChange = mock(ExecutableChange.class);
+        when(executableChange.getId()).thenReturn(id);
+        when(executableChange.getAction()).thenReturn(action);
+        when(executableChange.isAlreadyApplied()).thenReturn(action == ChangeAction.SKIP);
+        return executableChange;
     }
 
     private static ExecutionPlanResponse buildResponse(StageResponse... stages) {