Skip to content

Commit 7f1ee19

Browse files
dieppadieppa
committed
refactor: add StageRunBlock to PipelineRun
1 parent 26554fe commit 7f1ee19

15 files changed

Lines changed: 487 additions & 241 deletions

File tree

‎cloud/flamingock-cloud/src/main/java/io/flamingock/cloud/planner/CloudExecutionPlanner.java‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@
3333
import io.flamingock.internal.core.plan.ExecutionPlan;
3434
import io.flamingock.internal.core.plan.ExecutionPlanner;
3535
import io.flamingock.internal.core.external.store.lock.LockException;
36-
import io.flamingock.internal.core.pipeline.execution.ExecutableStage;
3736
import io.flamingock.internal.core.pipeline.loaded.stage.AbstractLoadedStage;
3837
import io.flamingock.internal.core.pipeline.run.PipelineRun;
3938
import io.flamingock.internal.util.log.FlamingockLoggerFactory;
@@ -103,8 +102,7 @@ public ExecutionPlan getNextExecution(PipelineRun pipelineRun) throws LockExcept
103102
}
104103

105104
if (response.isContinue()) {
106-
List<ExecutableStage> executableStages = CloudExecutionPlanMapper.getExecutableStages(response, loadedStages);
107-
return ExecutionPlan.CONTINUE(executableStages);
105+
return ExecutionPlan.CONTINUE();
108106

109107
} else if (response.isExecute()) {
110108
Lock lock = CloudLock.initialiseLocal(response.getLock(), coreConfiguration, runnerId, lockService, timeService);
@@ -132,8 +130,7 @@ public ExecutionPlan getNextExecution(PipelineRun pipelineRun) throws LockExcept
132130
);
133131

134132
} else if (response.isAbort()) {
135-
List<ExecutableStage> stages = CloudExecutionPlanMapper.getExecutableStages(response, loadedStages);
136-
return ExecutionPlan.ABORT(stages);
133+
return ExecutionPlan.ABORT();
137134

138135
} else {
139136
throw new RuntimeException("Unrecognized action from response. Not within(CONTINUE, EXECUTE, AWAIT, ABORT)");

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

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
import io.flamingock.internal.core.pipeline.execution.ExecutableStage;
3030
import io.flamingock.internal.core.pipeline.loaded.stage.AbstractLoadedStage;
3131
import io.flamingock.internal.core.pipeline.loaded.stage.DefaultLoadedStage;
32-
import io.flamingock.internal.core.plan.ExecutionPlan;
3332
import io.flamingock.core.cloud.changes._001__CloudChange1;
3433
import io.flamingock.core.cloud.changes._002__CloudChange2;
3534
import org.junit.jupiter.api.BeforeAll;
@@ -190,11 +189,10 @@ void shouldBuildAbortPlanFromAbortResponse() {
190189
);
191190

192191
List<ExecutableStage> result = CloudExecutionPlanMapper.getExecutableStages(response, loadedStages);
193-
ExecutionPlan plan = ExecutionPlan.ABORT(result);
194-
195-
assertTrue(plan.isAborted());
196-
assertFalse(plan.isExecutionRequired());
197192

193+
// CloudExecutionPlanMapper.getExecutableStages preserves change actions on the mapped
194+
// ExecutableStage regardless of the response action; verify that mapping directly.
195+
// ExecutionPlan.ABORT() no longer carries stages — that's a planner-control-flow concern.
198196
Map<String, ChangeAction> actions = result.get(0).getChanges().stream()
199197
.collect(Collectors.toMap(ExecutableChange::getId, ExecutableChange::getAction));
200198
assertEquals(ChangeAction.MANUAL_INTERVENTION, actions.get(change1.getId()));

‎cloud/flamingock-cloud/src/test/java/io/flamingock/cloud/planner/CloudExecutionPlannerTest.java‎

Lines changed: 5 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,6 @@
2626
import io.flamingock.cloud.api.vo.CloudTargetSystemAuditMarkType;
2727
import io.flamingock.cloud.lock.CloudLockService;
2828
import io.flamingock.cloud.planner.client.ExecutionPlannerClient;
29-
import io.flamingock.internal.common.core.error.FlamingockException;
30-
import io.flamingock.internal.common.core.recovery.ManualInterventionRequiredException;
3129
import io.flamingock.internal.common.core.targets.TargetSystemAuditMarkType;
3230
import io.flamingock.internal.core.change.loaded.AbstractLoadedChange;
3331
import io.flamingock.internal.core.change.loaded.LoadedChangeBuilder;
@@ -98,7 +96,7 @@ private CloudExecutionPlanner buildPlanner(List<TargetSystemAuditMarker> auditMa
9896
}
9997

10098
@Test
101-
@DisplayName("Should return ABORT plan when server returns ABORT with MANUAL_INTERVENTION changes")
99+
@DisplayName("Should return ABORT plan when server returns ABORT (regardless of change actions)")
102100
void shouldReturnAbortPlanWhenServerReturnsAbort() {
103101
CloudExecutionPlanner planner = buildPlanner(Collections.emptyList());
104102

@@ -114,16 +112,14 @@ void shouldReturnAbortPlanWhenServerReturnsAbort() {
114112

115113
ExecutionPlan plan = planner.getNextExecution(PipelineRun.of(stages));
116114

117-
// ExecutionPlan.validate() now only signals the abort itself; MI is a per-stage concern
118-
// enforced by ExecutableStage.validate() inside the operation lambda.
115+
// ABORT is a control-flow signal — the operation reads isAborted() and breaks the loop.
116+
// No exception flows from ExecutionPlan itself. MI per-stage is tested separately on
117+
// ExecutableStage.validate() (see core's ExecutableStageTest / AbstractPipelineTraverseOperationTest).
119118
assertTrue(plan.isAborted());
120-
assertThrows(FlamingockException.class, plan::validate);
121-
assertThrows(ManualInterventionRequiredException.class,
122-
() -> plan.getExecutableStages().get(0).validate());
123119
}
124120

125121
@Test
126-
@DisplayName("Should return ABORT plan that throws FlamingockException when server returns ABORT but no MI changes")
122+
@DisplayName("Should return ABORT plan when server returns ABORT (no MI changes)")
127123
void shouldReturnAbortPlanWhenServerReturnsAbortWithNoMIChanges() {
128124
CloudExecutionPlanner planner = buildPlanner(Collections.emptyList());
129125

@@ -140,7 +136,6 @@ void shouldReturnAbortPlanWhenServerReturnsAbortWithNoMIChanges() {
140136
ExecutionPlan plan = planner.getNextExecution(PipelineRun.of(stages));
141137

142138
assertTrue(plan.isAborted());
143-
assertThrows(io.flamingock.internal.common.core.error.FlamingockException.class, plan::validate);
144139
}
145140

146141
@Test

‎core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/response/data/ExecuteResponseData.java‎

Lines changed: 0 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -44,15 +44,6 @@ public class ExecuteResponseData {
4444
// Per-stage breakdown
4545
private List<StageResult> stages;
4646

47-
/**
48-
* @deprecated Per-stage error info is the authoritative carrier (see {@link StageResult} /
49-
* {@code StageState.getErrorInfo()}). The top-level field is kept temporarily to preserve the
50-
* JSON contract with the CLI; it is populated by {@code PipelineRun.toResponse()} from the
51-
* first failed stage's {@link ErrorInfo} and will be removed in a follow-up.
52-
*/
53-
@Deprecated
54-
private ErrorInfo error;
55-
5647
public ExecuteResponseData() {
5748
this.stages = new ArrayList<>();
5849
}
@@ -70,7 +61,6 @@ private ExecuteResponseData(Builder builder) {
7061
this.skippedChanges = builder.skippedChanges;
7162
this.failedChanges = builder.failedChanges;
7263
this.stages = builder.stages != null ? builder.stages : new ArrayList<>();
73-
this.error = builder.error;
7464
}
7565

7666
public ExecutionStatus getStatus() {
@@ -169,22 +159,6 @@ public void setStages(List<StageResult> stages) {
169159
this.stages = stages;
170160
}
171161

172-
/**
173-
* @deprecated See {@link #error}.
174-
*/
175-
@Deprecated
176-
public ErrorInfo getError() {
177-
return error;
178-
}
179-
180-
/**
181-
* @deprecated See {@link #error}.
182-
*/
183-
@Deprecated
184-
public void setError(ErrorInfo error) {
185-
this.error = error;
186-
}
187-
188162
public boolean isSuccess() {
189163
return status == ExecutionStatus.SUCCESS || status == ExecutionStatus.NO_CHANGES;
190164
}
@@ -210,7 +184,6 @@ public static class Builder {
210184
private int skippedChanges;
211185
private int failedChanges;
212186
private List<StageResult> stages = new ArrayList<>();
213-
private ErrorInfo error;
214187

215188
public Builder status(ExecutionStatus status) {
216189
this.status = status;
@@ -280,15 +253,6 @@ public Builder addStage(StageResult stage) {
280253
return this;
281254
}
282255

283-
/**
284-
* @deprecated See {@link ExecuteResponseData#error}.
285-
*/
286-
@Deprecated
287-
public Builder error(ErrorInfo error) {
288-
this.error = error;
289-
return this;
290-
}
291-
292256
public ExecuteResponseData build() {
293257
return new ExecuteResponseData(this);
294258
}

‎core/flamingock-core/src/main/java/io/flamingock/internal/core/operation/AbstractPipelineTraverseOperation.java‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import io.flamingock.internal.common.core.recovery.ManualInterventionRequiredException;
2020
import io.flamingock.internal.common.core.error.PendingChangesException;
2121
import io.flamingock.internal.common.core.response.data.ExecuteResponseData;
22+
import io.flamingock.internal.common.core.response.data.ExecutionStatus;
2223
import io.flamingock.internal.core.event.EventPublisher;
2324
import io.flamingock.internal.core.event.model.impl.PipelineCompletedEvent;
2425
import io.flamingock.internal.core.event.model.impl.PipelineFailedEvent;
@@ -121,7 +122,10 @@ private ExecuteResult execute(LoadedPipeline pipeline) {
121122

122123
do {
123124
try (ExecutionPlan execution = executionPlanner.getNextExecution(pipelineRun)) {
124-
execution.validate();
125+
if (execution.isAborted()) {
126+
logger.info("Pipeline execution aborted by the planner — earlier block has failures or another structural reason");
127+
break;
128+
}
125129

126130
if (!execution.isExecutionRequired()) {
127131
break;
@@ -134,7 +138,6 @@ private ExecuteResult execute(LoadedPipeline pipeline) {
134138
execution.applyOnEach((executionId, lock, executableStage) ->
135139
runStage(executionId, lock, executableStage, pipelineRun));
136140
} catch (LockException exception) {
137-
pipelineRun.markPipelineFailed(exception);
138141
pipelineLevelError = exception;
139142
if (throwExceptionIfCannotObtainLock) {
140143
logger.debug("Required process lock not acquired - ABORTING OPERATION", exception);
@@ -143,17 +146,19 @@ private ExecuteResult execute(LoadedPipeline pipeline) {
143146
throwPipelineLevelError = false;
144147
}
145148
break;
146-
} catch (Throwable exception) {
147-
pipelineRun.markPipelineFailed(exception);
149+
} catch (Throwable exception) {
148150
pipelineLevelError = exception;
149151
break;
150-
}//FlamingockException
152+
}
151153
} while (true);
152154

153155
pipelineRun.stop();
154156
ExecuteResponseData result = pipelineRun.toResponse();
155157

156158
if (pipelineLevelError != null) {
159+
// toResponse() derives status from per-stage state; override to FAILED so the
160+
// response reflects the pipeline-wide error carried in pipelineLevelError.
161+
result.setStatus(ExecutionStatus.FAILED);
157162
logger.debug("Error executing the process. ABORTED OPERATION", pipelineLevelError);
158163
eventPublisher.publish(new PipelineFailedEvent(toException(pipelineLevelError)));
159164
if (throwPipelineLevelError) {

‎core/flamingock-core/src/main/java/io/flamingock/internal/core/pipeline/run/PipelineRun.java‎

Lines changed: 7 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@
3636
import java.util.LinkedHashMap;
3737
import java.util.List;
3838
import java.util.Map;
39-
import java.util.Optional;
4039
import java.util.stream.Collectors;
4140

4241
public class PipelineRun {
@@ -98,7 +97,6 @@ private static StageType resolveType(AbstractLoadedStage stage) {
9897
private final Map<String, StageRun> byName;
9998
private Instant startedAt;
10099
private Instant stoppedAt;
101-
private ErrorInfo pipelineError;
102100

103101
private PipelineRun(List<StageRun> stageRuns, List<StageRunBlock> stageBlocks) {
104102
this.stageRuns = Collections.unmodifiableList(stageRuns);
@@ -205,19 +203,6 @@ public void markStageBlockedFromMI(String stageName, List<RecoveryIssue> issues)
205203
.build());
206204
}
207205

208-
/**
209-
* Pipeline-wide failure marker. Idempotent: first call wins.
210-
*/
211-
public void markPipelineFailed(Throwable cause) {
212-
if (this.pipelineError == null) {
213-
this.pipelineError = ErrorInfo.fromThrowable(cause, Collections.emptyList(), null);
214-
}
215-
}
216-
217-
public Optional<ErrorInfo> getPipelineError() {
218-
return Optional.ofNullable(pipelineError);
219-
}
220-
221206
private StageRun lookupOrThrow(String stageName) {
222207
StageRun run = byName.get(stageName);
223208
if (run == null) {
@@ -226,6 +211,12 @@ private StageRun lookupOrThrow(String stageName) {
226211
return run;
227212
}
228213

214+
/**
215+
* Builds the response data view derived purely from per-stage state. Status is FAILED iff any
216+
* stage failed, else SUCCESS. Pipeline-wide errors (e.g., LockException) that aren't reflected
217+
* in any stage state are signalled by the operation post-loop via
218+
* {@code ExecuteResponseData.setStatus(FAILED)} after {@code toResponse()} returns.
219+
*/
229220
public ExecuteResponseData toResponse() {
230221
List<StageResult> stages = new ArrayList<>();
231222
int totalStages = 0;
@@ -235,7 +226,6 @@ public ExecuteResponseData toResponse() {
235226
int appliedChanges = 0;
236227
int skippedChanges = 0;
237228
int failedChanges = 0;
238-
ErrorInfo error = null;
239229
boolean anyFailed = false;
240230

241231
for (StageRun stageRun : stageRuns) {
@@ -249,9 +239,6 @@ public ExecuteResponseData toResponse() {
249239
} else if (state.isFailed()) {
250240
failedStages++;
251241
anyFailed = true;
252-
if (error == null) {
253-
error = state.getErrorInfo().orElse(null);
254-
}
255242
}
256243

257244
if (stageResult.getChanges() != null) {
@@ -268,11 +255,7 @@ public ExecuteResponseData toResponse() {
268255
}
269256
}
270257

271-
if (error == null) {
272-
error = this.pipelineError;
273-
}
274-
boolean pipelineFailed = anyFailed || this.pipelineError != null;
275-
ExecutionStatus status = pipelineFailed ? ExecutionStatus.FAILED : ExecutionStatus.SUCCESS;
258+
ExecutionStatus status = anyFailed ? ExecutionStatus.FAILED : ExecutionStatus.SUCCESS;
276259
long durationMs = (startedAt != null && stoppedAt != null)
277260
? Duration.between(startedAt, stoppedAt).toMillis()
278261
: 0L;
@@ -290,7 +273,6 @@ public ExecuteResponseData toResponse() {
290273
.skippedChanges(skippedChanges)
291274
.failedChanges(failedChanges)
292275
.stages(stages)
293-
.error(error)
294276
.build();
295277
}
296278
}

‎core/flamingock-core/src/main/java/io/flamingock/internal/core/plan/ExecutionPlan.java‎

Lines changed: 22 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -18,25 +18,41 @@
1818
import io.flamingock.internal.util.TriConsumer;
1919
import io.flamingock.internal.core.external.store.lock.Lock;
2020
import io.flamingock.internal.core.pipeline.execution.ExecutableStage;
21-
import io.flamingock.internal.common.core.error.FlamingockException;
2221

22+
import java.util.Collections;
2323
import java.util.List;
2424

25+
/**
26+
* One iteration's planner verdict.
27+
*
28+
* <ul>
29+
* <li>{@link #newExecution(String, Lock, List)} — carries a list of stages to execute under the
30+
* given lock. {@link #isExecutionRequired()} reflects whether any of those stages still has
31+
* work pending.</li>
32+
* <li>{@link #CONTINUE()} — pipeline finished. Successfully. Nothing else to do.</li>
33+
* <li>{@link #ABORT()} — stop early. Something went wrong (e.g., an earlier block failed and
34+
* its dependents cannot proceed). The operation reads {@link #isAborted()} to break out of
35+
* the run loop.</li>
36+
* </ul>
37+
*
38+
* <p>{@code CONTINUE} and {@code ABORT} carry no stages — the run-loop only needs the verdict.
39+
* The pipeline-level state lives in {@code PipelineRun}; block-aware queries should read
40+
* {@code PipelineRun.getStageBlocks()} directly.
41+
*/
2542
public class ExecutionPlan implements AutoCloseable {
2643

27-
2844
public static ExecutionPlan newExecution(String executionId,
2945
Lock lock,
3046
List<ExecutableStage> stages) {
3147
return new ExecutionPlan(executionId, lock, stages);
3248
}
3349

34-
public static ExecutionPlan CONTINUE(List<ExecutableStage> stages) {
35-
return new ExecutionPlan(false, stages);
50+
public static ExecutionPlan CONTINUE() {
51+
return new ExecutionPlan(false, Collections.emptyList());
3652
}
3753

38-
public static ExecutionPlan ABORT(List<ExecutableStage> stages) {
39-
return new ExecutionPlan(true, stages);
54+
public static ExecutionPlan ABORT() {
55+
return new ExecutionPlan(true, Collections.emptyList());
4056
}
4157

4258
private final String executionId;
@@ -80,20 +96,6 @@ public void applyOnEach(TriConsumer<String, Lock, ExecutableStage> consumer) {
8096
}
8197
}
8298

83-
/**
84-
* Validates the execution plan. If the planner decided to abort the run for reasons beyond
85-
* individual change state, throws {@link FlamingockException}.
86-
*
87-
* <p>Manual-intervention validation is no longer performed here: it is per-stage and lives in
88-
* {@code ExecutableStage.validate()}, called inside the operation lambda so a single stage's
89-
* MI state never aborts the rest of the pipeline.
90-
*/
91-
public void validate() {
92-
if (aborted) {
93-
throw new FlamingockException("Execution aborted by the execution planner");
94-
}
95-
}
96-
9799
@Override
98100
public void close() {
99101
if (lock != null) {

0 commit comments

Comments
 (0)