svc-hall/svc-hall-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Preserve active leases and align flow recovery rounds

Browse Source 
Local review candidate only. Explicitly selected inherited prerequisites and prior repair patches; unrelated WIP and independent MQ fixes are excluded. No push or deployment authorized.
This commit is contained in:
chenxing
2026-10-23 10:51:40 +08:00
parent 8edccd5514
commit a4d220362e
18 changed files with 766 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/test/java/com/yunhai/svchall/service/impl/FlowCenterClientFacadeTest.java
+22
View File
@@ -287,4 +287,26 @@ class FlowCenterClientFacadeTest {
() -> c.listFlows(3L));
assertTrue(ex.getMessage().contains("流程中心未接入"), ex.getMessage());
}
@Test
void v2ResultRetainsBranchRoundAndRealTerminalStatus() {
com.yunhai.svchall.model.FcApplyModels.FlowOutbound outbound = new com.yunhai.svchall.model.FcApplyModels.FlowOutbound();
outbound.setBizId("42"); outbound.setRequestToken("token-1");
when(openCall.post(eq(TARGET), eq("/oapi/flow/apply/v2"), anyMap(), any(), eq(com.fasterxml.jackson.databind.JsonNode.class)))
.thenReturn(objectMapper.createObjectNode().put("instanceId", 555).put("duplicated", true)
.put("branch", "SAME_REQUEST").put("applyRound", 1).put("status", 30).put("version", 4));
FlowClient.ApplyResult result = client().applyWithIdentity(3L, outbound);
assertEquals("SAME_REQUEST", result.getBranch()); assertEquals(Integer.valueOf(1), result.getApplyRound());
assertEquals(Integer.valueOf(30), result.getStatus()); assertEquals(Integer.valueOf(4), result.getVersion());
assertEquals(Long.valueOf(555), result.getInstanceId());
}
@Test
void v2DoesNotTruncateFractionalRound() {
com.yunhai.svchall.model.FcApplyModels.FlowOutbound outbound = new com.yunhai.svchall.model.FcApplyModels.FlowOutbound();
outbound.setBizId("42"); outbound.setRequestToken("token-1");
when(openCall.post(eq(TARGET), eq("/oapi/flow/apply/v2"), anyMap(), any(), eq(com.fasterxml.jackson.databind.JsonNode.class)))
.thenReturn(objectMapper.createObjectNode().put("instanceId", 555).put("branch", "RESTARTED").put("applyRound", 2.5));
assertThrows(com.yunhai.platform.apistd.BizException.class, () -> client().applyWithIdentity(3L, outbound));
}
}
src/test/java/com/yunhai/svchall/service/impl/ServiceApplyFlowSubmitTest.java
+22
View File
@@ -299,4 +299,26 @@ class ServiceApplyFlowSubmitTest {
assertEquals(ApplyFlowStatusEnum.REJECTED.getCode(), apply.getFlowStatus().intValue(),
"出站失败保持原态，可再次重提");
}
@Test
void productionResubmitUsesCommandContractWithoutBlindApprovingWrite() {
apply.setFlowStatus(30); apply.setFcIntentId(1);
when(applyRepository.findByIdAndSchId(42, 1L)).thenReturn(Optional.of(apply));
FlowApplyRecoveryService commands = mock(FlowApplyRecoveryService.class);
service.setFlowCommandsForTest(commands);
service.resubmit(user(9, 1), 42);
verify(commands).requestExecution(42, true);
verify(flowClient, never()).apply(anyLong(), anyString(), anyString(), anyString(), anyString(), any(), any(), anyInt(), anyString(), any());
verify(applyRepository, never()).markResubmitted(anyInt(), anyLong(), any());
assertEquals(30, apply.getFlowStatus().intValue());
}
@Test
void guardedApplyCannotFallBackToLegacyWhenCommandServiceMissing() {
apply.setFlowStatus(30); apply.setFcIntentId(1);
when(applyRepository.findByIdAndSchId(42, 1L)).thenReturn(Optional.of(apply));
assertThrows(BizException.class, () -> service.resubmit(user(9, 1), 42));
verify(flowClient, never()).apply(anyLong(), anyString(), anyString(), anyString(), anyString(), any(), any(), anyInt(), anyString(), any());
assertEquals(30, apply.getFlowStatus().intValue());
}
}
src/test/java/com/yunhai/svchall/service/impl/ShAnswerFlowTest.java
+128 -4
View File
@@ -734,6 +734,8 @@ class ShAnswerFlowTest {
remote.put("instanceId", 4321L);
remote.put("status", 30);
remote.put("version", 2);
remote.put("applyRound", 1);
remote.put("requestToken", command.getRequestToken());
when(rh.flowClient.getInstanceByBiz(eq(SCH), anyString(), anyString())).thenReturn(remote);
rh.service.runCommand(command.getId());
@@ -743,7 +745,9 @@ class ShAnswerFlowTest {
assertThat(command.getState()).isEqualTo(FlowApplyCommand.STATE_DONE);
assertThat(command.getRemoteStatus()).isEqualTo(30);
verify(rh.applies).mirrorRemoteState(eq(42), eq(4321L), eq(30), eq(2), any(LocalDateTime.class));
assertThat(rh.mirror.getFlowInstanceId()).isEqualTo(4321L);
assertThat(rh.mirror.getFlowStatus()).isEqualTo(30);
assertThat(rh.mirror.getLastInstanceVersion()).isEqualTo(2);
verify(rh.flowClient, times(1)).applyWithIdentity(anyLong(), any());
verify(rh.applies, never()).markApproving(anyInt(), anyLong(), any(LocalDateTime.class));
}
@@ -761,7 +765,7 @@ class ShAnswerFlowTest {
assertThat(command.getState()).isEqualTo(FlowApplyCommand.STATE_UNKNOWN);
// 再桩必须走 doReturn：when(mock.apply(...)) 会真调用一次并把上一个 thenThrow 抛出来
org.mockito.Mockito.doReturn(new FlowClient.ApplyResult(555L, false))
org.mockito.Mockito.doReturn(new FlowClient.ApplyResult(555L, false, "CREATED", 1, 10, 1))
.when(rh.flowClient).applyWithIdentity(anyLong(), any());
rh.service.runCommand(command.getId());
@@ -771,7 +775,9 @@ class ShAnswerFlowTest {
verify(rh.flowClient, times(2)).applyWithIdentity(eq(SCH), sent.capture());
assertThat(sent.getAllValues()).allSatisfy(outbound ->
assertThat(outbound.getRequestToken()).isEqualTo(command.getRequestToken()));
verify(rh.applies).markApproving(eq(42), eq(555L), any(LocalDateTime.class));
assertThat(rh.mirror.getFlowStatus()).isEqualTo(10);
assertThat(rh.mirror.getFlowInstanceId()).isEqualTo(555L);
assertThat(sent.getAllValues().get(1).getRetryOnly()).isTrue();
}
@Test
@@ -825,7 +831,7 @@ class ShAnswerFlowTest {
command.setNextRetryAt(LocalDateTime.now().minusMinutes(1));
when(rh.commands.findDue(any(LocalDateTime.class), any()))
.thenReturn(Collections.singletonList(command));
org.mockito.Mockito.doReturn(new FlowClient.ApplyResult(777L, true))
org.mockito.Mockito.doReturn(new FlowClient.ApplyResult(777L, true, "SAME_REQUEST", 1, 30, 2))
.when(rh.flowClient).applyWithIdentity(anyLong(), any());
assertThat(rh.service.recoverDue()).isEqualTo(1);
@@ -833,6 +839,121 @@ class ShAnswerFlowTest {
assertThat(command.getSettledAt()).isNotNull();
}
@org.junit.jupiter.params.ParameterizedTest
@org.junit.jupiter.params.provider.CsvSource({
"CREATED,1,10,1", "IDEMPOTENT,1,20,2", "SAME_REQUEST,1,30,1",
"STALE_ROUND,3,40,2", "REPLAY_ONLY,1,30,2", "RESTARTED,2,10,2",
"CONCURRENT_CREATED,1,20,1"})
void allFlBranchesPreserveRealStatusAndAcceptedRound(String branch, int remoteRound, int status, int attempt) throws Exception {
RecoveryHarness rh = recoveryHarness();
FlowApplyCommand command = rh.pending(); command.setAttempt(attempt);
rh.mirror.setFlowStatus("CREATED".equals(branch) ? 0 : 30); rh.mirror.setLastInstanceVersion(1);
rh.mirror.setFlowResultJson("{\"applyRound\":1,\"result\":\"OLD_REJECTED\"}");
when(rh.flowClient.applyWithIdentity(anyLong(), any())).thenReturn(
new FlowClient.ApplyResult(555L, true, branch, remoteRound, status, 3));
rh.service.runCommand(command.getId());
assertThat(command.getState()).isEqualTo(FlowApplyCommand.STATE_DONE);
assertThat(command.getRemoteStatus()).isEqualTo(status);
assertThat(command.getAttempt()).isEqualTo(attempt); // 请求历史不可冒充远端实际轮次。
assertThat(rh.mirror.getFlowStatus()).isEqualTo(status);
assertThat(mapper.readTree(command.getPayloadJson()).path("_flOutcome").path("branch").asText()).isEqualTo(branch);
assertThat(mapper.readTree(command.getPayloadJson()).path("_flOutcome").path("applyRound").asInt()).isEqualTo(remoteRound);
rh.service.runCommand(command.getId());
verify(rh.flowClient, times(1)).applyWithIdentity(anyLong(), any());
}
@Test
void staleRoundCannotTurnLocalTerminalBackIntoApprovingEvenWithHigherVersion() {
RecoveryHarness rh = recoveryHarness();
FlowApplyCommand command = rh.pending(); command.setAttempt(2);
rh.mirror.setFlowStatus(40); rh.mirror.setLastInstanceVersion(3);
rh.mirror.setFlowResultJson("{\"applyRound\":4,\"result\":\"WITHDRAWN\"}");
when(rh.flowClient.applyWithIdentity(anyLong(), any())).thenReturn(
new FlowClient.ApplyResult(555L, true, "STALE_ROUND", 3, 10, 99));
rh.service.runCommand(command.getId());
assertThat(rh.mirror.getFlowStatus()).isEqualTo(40);
assertThat(rh.mirror.getLastInstanceVersion()).isEqualTo(3);
assertThat(rh.mirror.getFlowResultJson()).contains("WITHDRAWN");
assertThat(command.getRemoteStatus()).isEqualTo(10); // 记录远端实际响应，不把本地保护结果冒充远端状态。
}
@Test
void restartedResponseCannotOverwriteQuickNewRoundTerminalCallback() {
RecoveryHarness rh = recoveryHarness();
FlowApplyCommand command = rh.pending(); command.setAttempt(2);
rh.mirror.setFlowInstanceId(555L); rh.mirror.setFlowStatus(20); rh.mirror.setLastInstanceVersion(5);
rh.mirror.setFlowResultJson("{\"applyRound\":2,\"result\":\"APPROVED\"}");
when(rh.flowClient.applyWithIdentity(anyLong(), any())).thenReturn(
new FlowClient.ApplyResult(555L, true, "RESTARTED", 2, 10, 4));
rh.service.runCommand(command.getId());
assertThat(rh.mirror.getFlowStatus()).isEqualTo(20);
assertThat(rh.mirror.getFlowResultJson()).contains("APPROVED");
}
@Test
void unknownBranchRemainsUnknownAndNeverWritesApproving() {
RecoveryHarness rh = recoveryHarness(); FlowApplyCommand command = rh.pending();
when(rh.flowClient.applyWithIdentity(anyLong(), any())).thenReturn(new FlowClient.ApplyResult(555L, false));
rh.service.runCommand(command.getId());
assertThat(command.getState()).isEqualTo(FlowApplyCommand.STATE_UNKNOWN);
assertThat(rh.mirror.getFlowStatus()).isZero();
verify(rh.applies, never()).save(any());
}
@Test
void instanceQueryWithForeignTokenDoesNotSettleOrMirror() {
RecoveryHarness rh = recoveryHarness(); FlowApplyCommand command = rh.pending();
command.setState(FlowApplyCommand.STATE_UNKNOWN);
Map<String, Object> remote = new HashMap<>();
remote.put("instanceId", 555L); remote.put("status", 30); remote.put("version", 2);
remote.put("applyRound", 1); remote.put("requestToken", "not-our-command");
when(rh.flowClient.getInstanceByBiz(anyLong(), anyString(), anyString())).thenReturn(remote);
rh.service.runCommand(command.getId());
assertThat(command.getState()).isEqualTo(FlowApplyCommand.STATE_UNKNOWN);
rh.service.runCommand(command.getId());
assertThat(rh.mirror.getFlowStatus()).isZero();
verify(rh.flowClient, never()).applyWithIdentity(anyLong(), any());
}
@Test
void delayedOlderResponseCannotOverwriteNewerCommandOrTerminalMirror() {
RecoveryHarness rh = recoveryHarness(); FlowApplyCommand command = rh.pending();
java.util.concurrent.atomic.AtomicInteger calls = new java.util.concurrent.atomic.AtomicInteger();
when(rh.flowClient.applyWithIdentity(anyLong(), any())).thenAnswer(call -> {
if (calls.incrementAndGet() == 1) {
rh.service.runCommand(command.getId()); // 第二发送者先收到最新终态。
return new FlowClient.ApplyResult(555L, false, "CREATED", 1, 10, 4);
}
return new FlowClient.ApplyResult(555L, true, "SAME_REQUEST", 1, 20, 5);
});
rh.service.runCommand(command.getId());
assertThat(rh.mirror.getFlowStatus()).isEqualTo(20);
assertThat(command.getRemoteStatus()).isEqualTo(20);
assertThat(command.getRemoteVersion()).isEqualTo(5);
assertThat(command.getState()).isEqualTo(FlowApplyCommand.STATE_DONE);
}
@Test
void authorizedResubmitAllocatesSecondRoundAndDoubleClickReusesUnknownCommand() {
RecoveryHarness rh = recoveryHarness(); rh.pending();
rh.mirror.setFlowStatus(30); rh.mirror.setFlowResultJson("{\"applyRound\":1}");
when(rh.flowClient.applyWithIdentity(anyLong(), any())).thenThrow(new IllegalStateException("response lost"));
rh.service.requestExecution(42, true);
assertThat(rh.created).hasSize(2);
FlowApplyCommand second = rh.created.get(1);
assertThat(second.getAttempt()).isEqualTo(2);
assertThat(second.getState()).isEqualTo(FlowApplyCommand.STATE_UNKNOWN);
String token = second.getRequestToken();
when(rh.flowClient.getInstanceByBiz(anyLong(), anyString(), anyString())).thenThrow(new IllegalStateException("probe unavailable"));
rh.service.requestExecution(42, true);
rh.service.requestExecution(42, true); // 再次查询失败仍 UNKNOWN，不能降成 FAILED 后盲目重发。
assertThat(second.getState()).isEqualTo(FlowApplyCommand.STATE_UNKNOWN);
assertThat(rh.created).hasSize(2);
assertThat(second.getRequestToken()).isEqualTo(token);
assertThat(rh.mirror.getFlowStatus()).isEqualTo(30);
verify(rh.flowClient, times(1)).applyWithIdentity(anyLong(), any());
}
// ==================== 夹具 ====================
private Snapshot snapshot(String valuesJson) {
@@ -1106,6 +1227,8 @@ class ShAnswerFlowTest {
RecoveryHarness rh = new RecoveryHarness();
rh.commands = mock(FlowApplyCommandRepository.class);
rh.applies = mock(ServiceApplyRepository.class);
rh.mirror = applyRow();
when(rh.applies.findByIdForUpdate(42)).thenReturn(Optional.of(rh.mirror));
rh.flowClient = mock(FlowClient.class);
when(rh.flowClient.enabled()).thenReturn(true);
when(rh.commands.saveAndFlush(any())).thenAnswer(call -> {
@@ -1174,6 +1297,7 @@ class ShAnswerFlowTest {
private final class RecoveryHarness {
FlowApplyCommandRepository commands;
ServiceApplyRepository applies;
ServiceApply mirror;
FlowClient flowClient;
FlowApplyRecoveryService service;
final List<FlowApplyCommand> created = new ArrayList<>();
src/test/java/com/yunhai/svchall/service/impl/ShReconcileTaskTest.java
+131 -3
View File
@@ -115,7 +115,13 @@ class ShReconcileTaskTest {
ReflectionTestUtils.setField(worker, "backoffBaseSeconds", 60);
ReflectionTestUtils.setField(worker, "backoffCapSeconds", 1800);
when(formSyncClient.enabled()).thenReturn(true);
when(formSyncClient.submitResult(anyString(), any(ResultReport.class))).thenReturn(acceptedAck());
when(formSyncClient.submitResult(anyString(), any(ResultReport.class))).thenAnswer(call -> {
ResultReport report = call.getArgument(1);
ResultAck ack = acceptedAck();
ack.setAttempt(report.getAttempt()); ack.setReportAttempt(report.getAttempt());
ack.setResultState(report.getState());
return ack;
});
}
// ==================== 接收与执行解耦 ====================
@@ -210,7 +216,7 @@ class ShReconcileTaskTest {
}
@Test
@DisplayName("仍在跑且租约有效时，新 attempt 只刷新身份：不打断正在执行的这一轮")
@DisplayName("有效租约中保留执行 attempt、状态与进度；过期后才接纳更大 attempt")
void redispatchWhileLeaseLiveDoesNotInterruptRunningRound() {
AnswerReconcileWork running = work(AnswerReconcileWork.STATE_RUNNING, 1);
running.setLeaseUntil(LocalDateTime.now().plusMinutes(10));
@@ -221,9 +227,89 @@ class ShReconcileTaskTest {
AnswerReconcileWork accepted = worker.accept(instruction(2), FC);
assertThat(accepted.getState()).isEqualTo(AnswerReconcileWork.STATE_RUNNING);
assertThat(accepted.getAttempt()).isEqualTo(2);
assertThat(accepted.getAttempt()).isEqualTo(1);
assertThat(accepted.getWatermark()).isEqualTo(ANSWER_A);
assertThat(accepted.getScannedRows()).isEqualTo(4);
assertThat(accepted.getLeaseUntil()).isEqualTo(running.getLeaseUntil());
verify(workRepository, never()).save(any(AnswerReconcileWork.class));
running.setLeaseUntil(LocalDateTime.now().minusSeconds(1));
AnswerReconcileWork resumed = worker.accept(instruction(2), FC);
assertThat(resumed.getAttempt()).isEqualTo(2);
assertThat(resumed.getState()).isEqualTo(AnswerReconcileWork.STATE_QUEUED);
assertThat(resumed.getWatermark()).isEqualTo(ANSWER_A);
assertThat(resumed.getScannedRows()).isEqualTo(4);
}
@Test
void leaseBoundaryAndDuplicateClaimAreCheckedUnderLock() {
java.time.Clock fixed = java.time.Clock.fixed(java.time.Instant.parse("2026-10-23T00:00:00Z"), java.time.ZoneOffset.UTC);
ReflectionTestUtils.setField(worker, "clock", fixed);
LocalDateTime now = LocalDateTime.now(fixed);
AnswerReconcileWork work = givenQueuedWork();
work.setState(AnswerReconcileWork.STATE_RUNNING);
work.setLeaseUntil(now.plusNanos(1));
assertThat(worker.executeQuietly(work.getId())).isZero();
verify(formSyncClient, never()).fetchManifestPage(anyString(), any(), anyInt());
work.setLeaseUntil(now); // <= now 与 SQL stale-running 选择器一致。
stubStreams(Collections.emptyList(), Collections.emptyList(), true, 200);
when(formSyncClient.fetchManifestPage(anyString(), any(), anyInt())).thenAnswer(call -> {
assertThat(worker.executeQuietly(work.getId())).isZero(); // 第一领取已更新租约，重复候选不扫描。
ManifestPageView page = new ManifestPageView();
page.setTaskId(TASK_ID); page.setRows(Collections.emptyList()); page.setCompleted(true);
return page;
});
assertThat(worker.executeQuietly(work.getId())).isEqualTo(1);
verify(formSyncClient, times(1)).fetchManifestPage(anyString(), any(), anyInt());
assertThat(work.getResumed()).isEqualTo(1);
}
@Test
void staleDueListCannotBypassBackoffOrTerminalState() {
AnswerReconcileWork work = givenQueuedWork();
work.setNextRetryAt(LocalDateTime.now().plusHours(1));
assertThat(worker.executeQuietly(work.getId())).isZero();
work.setNextRetryAt(null);
work.setState(AnswerReconcileWork.STATE_COMPLETED);
assertThat(worker.executeQuietly(work.getId())).isZero();
verify(formSyncClient, never()).fetchManifestPage(anyString(), any(), anyInt());
}
@Test
void redeliveryDuringScanCannotInvalidateCheckpointOrStartSecondScan() {
AnswerReconcileWork work = givenQueuedWork();
stubStreams(Collections.emptyList(), Collections.emptyList(), true, 200);
when(formSyncClient.fetchManifestPage(anyString(), any(), anyInt())).thenAnswer(call -> {
LocalDateTime lease = work.getLeaseUntil();
AnswerReconcileWork duplicate = worker.accept(instruction(2), FC);
assertThat(duplicate.getAttempt()).isEqualTo(1);
assertThat(duplicate.getState()).isEqualTo(AnswerReconcileWork.STATE_RUNNING);
assertThat(duplicate.getLeaseUntil()).isEqualTo(lease);
assertThat(worker.executeQuietly(work.getId())).isZero();
ManifestPageView page = new ManifestPageView();
page.setTaskId(TASK_ID); page.setRows(Collections.emptyList()); page.setCompleted(true);
return page;
});
assertThat(worker.executeQuietly(work.getId())).isEqualTo(1);
assertThat(work.getState()).isEqualTo(AnswerReconcileWork.STATE_COMPLETED);
verify(formSyncClient, times(1)).fetchManifestPage(anyString(), any(), anyInt());
}
@Test
void expiredOldWorkerCannotReleaseNewOwnersLeaseOnCheckpointConflict() {
AnswerReconcileWork old = givenQueuedWork(); old.setWorkVersion(1L);
AnswerReconcileWork newer = work(AnswerReconcileWork.STATE_RUNNING, 1);
newer.setWorkVersion(2L); newer.setLeaseUntil(LocalDateTime.now().plusMinutes(10));
LocalDateTime lease = newer.getLeaseUntil();
when(workRepository.findById(old.getId())).thenReturn(Optional.of(newer));
when(formSyncClient.fetchManifestPage(anyString(), any(), anyInt()))
.thenThrow(new OptimisticLockingFailureException("old checkpoint generation"));
worker.executeQuietly(old.getId());
assertThat(newer.getState()).isEqualTo(AnswerReconcileWork.STATE_RUNNING);
assertThat(newer.getLeaseUntil()).isEqualTo(lease);
assertThat(newer.getWorkVersion()).isEqualTo(2L);
verify(workRepository, never()).save(newer);
}
@Test
@@ -742,6 +828,45 @@ class ShReconcileTaskTest {
assertThat(work.getReportedAttempt()).isEqualTo(work.getAttempt());
}
@Test
void explicitNewerCenterTerminalStopsRetryWithoutPretendingSuccess() {
AnswerReconcileWork work = givenQueuedWork();
stubStreams(Collections.emptyList(), Collections.emptyList(), true, 200);
ResultAck ack = acceptedAck();
ack.setDecision("CENTER_TERMINAL"); ack.setAttempt(3);
ack.setStaleAttempt(true); ack.setTaskState(AnswerReconcileWork.STATE_FAILED);
when(formSyncClient.submitResult(anyString(), any())).thenReturn(ack);
worker.executeQuietly(work.getId());
assertThat(work.getState()).isEqualTo(AnswerReconcileWork.STATE_FAILED);
assertThat(work.getNextRetryAt()).isNull();
assertThat(work.getLeaseUntil()).isNull();
}
@Test
void duplicateFailureCannotBeReportedAsCompleted() {
AnswerReconcileWork work = givenQueuedWork();
stubStreams(Collections.emptyList(), Collections.emptyList(), true, 200);
ResultAck ack = acceptedAck(); ack.setDecision("CURRENT_DUPLICATE");
ack.setResultState(AnswerReconcileWork.STATE_FAILED);
when(formSyncClient.submitResult(anyString(), any())).thenReturn(ack);
worker.executeQuietly(work.getId());
assertThat(work.getState()).isEqualTo(AnswerReconcileWork.STATE_FAILED);
}
@Test
void missingOrMismatchedAckRemainsReportOnlyRetryable() {
AnswerReconcileWork work = givenQueuedWork();
stubStreams(Collections.emptyList(), Collections.emptyList(), true, 200);
ResultAck ack = acceptedAck(); ack.setReportAttempt(99);
when(formSyncClient.submitResult(anyString(), any())).thenReturn(ack);
worker.executeQuietly(work.getId());
assertThat(work.getState()).isEqualTo(AnswerReconcileWork.STATE_QUEUED);
assertThat(work.getPendingReportState()).isEqualTo(AnswerReconcileWork.STATE_COMPLETED);
assertThat(work.getReportedState()).isNull();
assertThat(work.getNextRetryAt()).isNotNull();
assertThat(work.getLeaseUntil()).isNull();
}
@Test
@DisplayName("进度回报按轮刷新中心租约，RUNNING 不落本地终态")
void progressReportsRefreshLeaseWithoutTerminalState() {
@@ -980,6 +1105,9 @@ class ShReconcileTaskTest {
ResultAck ack = new ResultAck();
ack.setTaskId(TASK_ID);
ack.setAccepted(Boolean.TRUE);
ack.setAttempt(1);
ack.setReportAttempt(1);
ack.setDecision("CURRENT_ACCEPTED");
ack.setStaleAttempt(Boolean.FALSE);
ack.setTaskState(AnswerReconcileWork.STATE_COMPLETED);
return ack;
src/test/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandlerTest.java
+67 -2
View File
@@ -31,6 +31,7 @@ class SvcFlowCallbackHandlerTest {
private ServiceApplyRepository applyRepository;
private SvcFlowCallbackHandler handler;
private ServiceApply apply;
private com.yunhai.svchall.repository.FlowApplyCommandRepository commands;
@BeforeEach
void setUp() {
@@ -40,10 +41,21 @@ class SvcFlowCallbackHandlerTest {
apply = new ServiceApply();
apply.setId(7);
apply.setSchId(1L);
apply.setFlowGuid("fg-1");
apply.setFlowStatus(10);
apply.setLastInstanceVersion(1);
when(applyRepository.findByIdForUpdate(7)).thenReturn(Optional.of(apply));
when(applyRepository.save(any(ServiceApply.class))).thenAnswer(inv -> inv.getArgument(0));
commands = mock(com.yunhai.svchall.repository.FlowApplyCommandRepository.class);
org.springframework.test.util.ReflectionTestUtils.setField(handler, "commandRepository", commands);
for (int round = 1; round <= 2; round++) {
com.yunhai.svchall.entity.FlowApplyCommand c = new com.yunhai.svchall.entity.FlowApplyCommand();
c.setApplyId(7); c.setSchId(1L); c.setAttempt(round); c.setRequestToken("token-" + round);
c.setFlowGuid("fg-1"); c.setSentAt(java.time.LocalDateTime.now()); c.setPayloadJson("{}");
when(commands.findByApplyIdAndAttempt(7, round)).thenReturn(Optional.of(c));
}
com.yunhai.svchall.entity.FlowApplyCommand first = commands.findByApplyIdAndAttempt(7, 1).get();
when(commands.findByApplyIdOrderByIdDesc(7)).thenReturn(java.util.Collections.singletonList(first));
}
private FlowCallbackEvent event(String eventId, String event, Integer version) {
@@ -60,6 +72,7 @@ class SvcFlowCallbackHandlerTest {
e.setOperatorUserId(201);
e.setOperatorName("张三");
Map<String, Object> data = new HashMap<>();
data.put("applyRound", 1); data.put("requestToken", "token-1");
e.setData(data);
return e;
}
@@ -202,7 +215,7 @@ class SvcFlowCallbackHandlerTest {
apply.setFlowResultJson("{\"event\":\"flow.rejected\"}");
apply.setFinishTime(java.time.LocalDateTime.now());
apply.setLastInstanceVersion(3);
handler.onEvent(event("r1", "flow.submitted", 4));
handler.onEvent(nextRound(event("r1", "flow.submitted", 4)));
assertEquals(10, apply.getFlowStatus().intValue(), "新轮（版本更高）回调放行，回开为 10");
assertEquals(4, apply.getLastInstanceVersion().intValue());
assertNull(apply.getFinishTime(), "回开为在途时清残留 finishTime");
@@ -216,7 +229,7 @@ class SvcFlowCallbackHandlerTest {
apply.setFlowResultJson("{\"event\":\"flow.rejected\",\"opinion\":\"旧轮意见\"}");
apply.setFinishTime(java.time.LocalDateTime.now());
apply.setLastInstanceVersion(3);
FlowCallbackEvent e = event("r2", "flow.completed", 4);
FlowCallbackEvent e = nextRound(event("r2", "flow.completed", 4));
e.getData().put("result", "APPROVED");
handler.onEvent(e);
assertEquals(20, apply.getFlowStatus().intValue(), "新轮终态覆盖旧轮终态");
@@ -239,4 +252,56 @@ class SvcFlowCallbackHandlerTest {
assertEquals(30, apply.getFlowStatus().intValue(), "旧轮事件全部拒收");
assertEquals(frozen, apply.getFlowResultJson(), "旧轮结果不被覆盖");
}
private FlowCallbackEvent nextRound(FlowCallbackEvent e) {
e.getData().put("applyRound", 2); e.getData().put("requestToken", "token-2"); return e;
}
@Test
void higherVersionInSameRoundDoesNotProveRestart() {
apply.setFlowStatus(30); apply.setLastInstanceVersion(3);
apply.setFlowResultJson("{\"applyRound\":1,\"result\":\"REJECTED\"}");
handler.onEvent(event("old-high-version", "flow.submitted", 99));
assertEquals(30, apply.getFlowStatus().intValue());
assertEquals(3, apply.getLastInstanceVersion().intValue());
}
@Test
void oldRoundBacklogCannotOverwriteNewRoundTerminalEvenWithHigherVersion() {
apply.setFlowStatus(20); apply.setLastInstanceVersion(5);
apply.setFlowResultJson("{\"applyRound\":2,\"result\":\"APPROVED\"}");
String frozen = apply.getFlowResultJson();
handler.onEvent(event("old-round", "flow.submitted", 100));
handler.onEvent(event("old-round-terminal", "flow.rejected", 100));
assertEquals(20, apply.getFlowStatus().intValue()); assertEquals(frozen, apply.getFlowResultJson());
}
@Test
void tokenMismatchOrUnsentCommandCannotAuthorizeNewRound() {
apply.setFlowStatus(40); apply.setLastInstanceVersion(3);
FlowCallbackEvent bad = nextRound(event("wrong-token", "flow.submitted", 4));
bad.getData().put("requestToken", "foreign"); handler.onEvent(bad);
assertEquals(40, apply.getFlowStatus().intValue());
commands.findByApplyIdAndAttempt(7, 2).get().setSentAt(null);
handler.onEvent(nextRound(event("unsent", "flow.submitted", 5)));
assertEquals(40, apply.getFlowStatus().intValue());
}
@Test
void duplicateNewRoundCallbackKeepsTerminalAndResult() {
apply.setFlowStatus(30); apply.setLastInstanceVersion(3);
FlowCallbackEvent completed = nextRound(event("new-done", "flow.completed", 4));
handler.onEvent(completed); String frozen = apply.getFlowResultJson();
handler.onEvent(completed);
handler.onEvent(nextRound(event("late-task", "flow.task.created", 4)));
assertEquals(20, apply.getFlowStatus().intValue()); assertEquals(frozen, apply.getFlowResultJson());
}
@Test
void managedCommandCannotAdvanceFromCallbackWithoutRoundIdentity() {
apply.setFlowStatus(30); apply.setLastInstanceVersion(3);
FlowCallbackEvent e = event("missing-round", "flow.submitted", 99); e.getData().clear();
handler.onEvent(e);
assertEquals(30, apply.getFlowStatus().intValue()); assertEquals(3, apply.getLastInstanceVersion().intValue());
}
}