findByTaskUuid(String taskUuid);
@Lock(LockModeType.PESSIMISTIC_WRITE)
diff --git a/src/main/java/com/yunhai/svchall/repository/FlowApplyCommandRepository.java b/src/main/java/com/yunhai/svchall/repository/FlowApplyCommandRepository.java
index efb6a8d..8e9b9a7 100644
--- a/src/main/java/com/yunhai/svchall/repository/FlowApplyCommandRepository.java
+++ b/src/main/java/com/yunhai/svchall/repository/FlowApplyCommandRepository.java
@@ -15,9 +15,8 @@ import java.util.Optional;
/**
* 出站发起命令仓库（TASK_2 C.1/C.2）。
*
- * {@link #findByIdForUpdate} 与既有 {@code ServiceApplyRepository#findByIdForUpdate} 同构：
- * 调度线程池与恢复扫描可能同时盯上同一条命令，行锁把两者串起来，避免"两个线程各自以为
- * 自己是唯一发送者"而双发（FL 侧 uk 与 requestToken 是第二、第三道闸）。
+ * 短事务行锁用于读取和结算持久命令，不跨 HTTP 持有；同命令并发发送仍可能发生。
+ * 远端按 requestToken/attempt 幂等吸收，本地结算再锁命令并按远端轮次/版本防止旧响应倒写。
*/
public interface FlowApplyCommandRepository extends JpaRepository {
diff --git a/src/main/java/com/yunhai/svchall/service/FlowClient.java b/src/main/java/com/yunhai/svchall/service/FlowClient.java
index 095ee11..303852d 100644
--- a/src/main/java/com/yunhai/svchall/service/FlowClient.java
+++ b/src/main/java/com/yunhai/svchall/service/FlowClient.java
@@ -79,22 +79,29 @@ public interface FlowClient {
*/
byte[] diagramImage(Long schId, Long instanceId, String flowGuid, String mode);
- /** 代发起结果 */
+ /** v2 结果保留 FL 原始分支、轮次及真实实例状态；旧 apply 构造不伪造 v2 语义。 */
class ApplyResult {
private final Long instanceId;
private final boolean duplicated;
+ private final String branch;
+ private final Integer applyRound;
+ private final Integer status;
+ private final Integer version;
public ApplyResult(Long instanceId, boolean duplicated) {
- this.instanceId = instanceId;
- this.duplicated = duplicated;
+ this(instanceId, duplicated, null, null, null, null);
}
- public Long getInstanceId() {
- return instanceId;
- }
-
- public boolean isDuplicated() {
- return duplicated;
+ public ApplyResult(Long instanceId, boolean duplicated, String branch, Integer applyRound,
+ Integer status, Integer version) {
+ this.instanceId = instanceId; this.duplicated = duplicated; this.branch = branch;
+ this.applyRound = applyRound; this.status = status; this.version = version;
}
+ public Long getInstanceId() { return instanceId; }
+ public boolean isDuplicated() { return duplicated; }
+ public String getBranch() { return branch; }
+ public Integer getApplyRound() { return applyRound; }
+ public Integer getStatus() { return status; }
+ public Integer getVersion() { return version; }
}
}
diff --git a/src/main/java/com/yunhai/svchall/service/impl/AnswerReconcileWorker.java b/src/main/java/com/yunhai/svchall/service/impl/AnswerReconcileWorker.java
index df8c432..1446362 100644
--- a/src/main/java/com/yunhai/svchall/service/impl/AnswerReconcileWorker.java
+++ b/src/main/java/com/yunhai/svchall/service/impl/AnswerReconcileWorker.java
@@ -28,6 +28,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionTemplate;
import java.time.LocalDateTime;
+import java.time.Clock;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.LinkedHashMap;
@@ -59,6 +60,7 @@ import java.util.Map;
@Slf4j
@Service
public class AnswerReconcileWorker {
+ private Clock clock = Clock.systemDefaultZone();
private static final DateTimeFormatter TS = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss");
private static final int TASK_UUID_MAX = 36;
@@ -166,6 +168,13 @@ public class AnswerReconcileWorker {
if (attempt <= intOf(row.getAttempt())) {
return row; // Same/older instruction cannot reopen a terminal work row.
}
+ if (AnswerReconcileWork.STATE_RUNNING.equals(row.getState())
+ && row.getLeaseUntil() != null
+ && row.getLeaseUntil().isAfter(LocalDateTime.now(clock))) {
+ // 接收与领取使用同一行锁；有效执行轮不换 attempt，也不触碰 @Version。
+ // 否则已分离的扫描对象下次 checkpoint 会被重派写入制造的乐观锁冲突打断。
+ return row;
+ }
}
row.setAttempt(attempt);
row.setConsumerAppKey(instruction.getConsumerAppKey());
@@ -268,6 +277,7 @@ public class AnswerReconcileWorker {
transactionTemplate.executeWithoutResult(tx -> {
AnswerReconcileWork row = workRepository.findById(claimed.getId()).orElse(claimed);
if (!java.util.Objects.equals(row.getAttempt(), claimed.getAttempt())
+ || !java.util.Objects.equals(row.getWorkVersion(), claimed.getWorkVersion())
|| AnswerReconcileWork.isTerminal(row.getState())) {
return; // Old worker must not overwrite a newer instruction or terminal decision.
}
@@ -296,8 +306,15 @@ public class AnswerReconcileWorker {
if (row == null || AnswerReconcileWork.isTerminal(row.getState())) {
return null;
}
+ LocalDateTime now = LocalDateTime.now(clock);
boolean running = AnswerReconcileWork.STATE_RUNNING.equals(row.getState());
- if (running && row.getLeaseUntil() != null && row.getLeaseUntil().isAfter(LocalDateTime.now())) {
+ if (!running && !AnswerReconcileWork.STATE_QUEUED.equals(row.getState())) {
+ return null;
+ }
+ if (!running && row.getNextRetryAt() != null && row.getNextRetryAt().isAfter(now)) {
+ return null; // 候选列表可能已过时：领取时重新检查退避，防止双调度绕过重试间隔。
+ }
+ if (running && row.getLeaseUntil() != null && row.getLeaseUntil().isAfter(now)) {
return null;
}
if (running) {
@@ -305,7 +322,7 @@ public class AnswerReconcileWorker {
}
row.setState(AnswerReconcileWork.STATE_RUNNING);
row.setStartedAt(row.getStartedAt() == null ? LocalDateTime.now() : row.getStartedAt());
- row.setLeaseUntil(LocalDateTime.now().plusSeconds(leaseSeconds));
+ row.setLeaseUntil(now.plusSeconds(leaseSeconds));
row.setNextRetryAt(null);
return workRepository.save(row);
});
@@ -718,6 +735,8 @@ public class AnswerReconcileWorker {
int failures = intOf(context.work.getRetryCount()) + 1;
context.work.setRetryCount(failures);
boolean exhausted = failures >= Math.max(1, maxExecutionRetries);
+ context.work.setReportedState(null);
+ context.work.setReportedAttempt(null);
context.work.setPendingReportState(state);
context.work.setState(exhausted ? AnswerReconcileWork.STATE_FAILED : AnswerReconcileWork.STATE_QUEUED);
context.work.setLastErrorCode(FormSyncModels.ERR_CLIENT_UNAVAILABLE);
@@ -758,25 +777,49 @@ public class AnswerReconcileWorker {
}
private void applyAck(AnswerReconcileWork work, ResultAck ack, String state) {
+ if (ack == null || !Boolean.TRUE.equals(ack.getAccepted())
+ || !work.getTaskUuid().equals(ack.getTaskId())
+ || (ack.getReportAttempt() != null && !ack.getReportAttempt().equals(work.getAttempt()))) {
+ throw new IllegalStateException("INVALID_RESULT_ACK：未确认或回执身份/回报轮次不符");
+ }
work.setReportedAttempt(work.getAttempt());
work.setReportedState(state);
- if (ack != null && ack.getTaskState() != null && AnswerReconcileWork.isTerminal(ack.getTaskState())) {
+ String decision = ack.getDecision();
+ if ("CENTER_TERMINAL".equals(decision)) {
+ if (!AnswerReconcileWork.isTerminal(ack.getTaskState()) || ack.getAttempt() == null
+ || ack.getAttempt() < intOf(work.getAttempt())) {
+ throw new IllegalStateException("INVALID_RESULT_ACK：中心终态证明不完整或轮次陈旧");
+ }
work.setState(ack.getTaskState());
work.setFinishedAt(LocalDateTime.now());
work.setNextRetryAt(null);
work.setLeaseUntil(null);
return;
}
- if (ack != null && Boolean.TRUE.equals(ack.getStaleAttempt())) {
- // 中心判这次回报是旧 attempt：本地不覆盖已提交结论，等中心重派
+ if (Boolean.TRUE.equals(ack.getStaleAttempt()) || "STALE_RETRY".equals(decision)) {
+ // 诊断 taskState 并不确认本次旧轮；只有显式 CENTER_TERMINAL 才能停止重试。
work.setState(AnswerReconcileWork.STATE_QUEUED);
+ work.setLeaseUntil(null);
+ work.setFinishedAt(null);
work.setNextRetryAt(LocalDateTime.now().plusSeconds(backoffSeconds(1)));
work.setLastErrorCode("STALE_ATTEMPT");
return;
}
- work.setState(state);
- work.setFinishedAt(AnswerReconcileWork.STATE_RUNNING.equals(state) ? null : LocalDateTime.now());
- work.setLeaseUntil(AnswerReconcileWork.STATE_RUNNING.equals(state)
+ if (decision != null && !"CURRENT_ACCEPTED".equals(decision) && !"CURRENT_DUPLICATE".equals(decision)) {
+ throw new IllegalStateException("INVALID_RESULT_ACK：未知回执决策");
+ }
+ if (decision != null && !java.util.Objects.equals(ack.getAttempt(), work.getAttempt())) {
+ throw new IllegalStateException("INVALID_RESULT_ACK：当前回执轮次不符");
+ }
+ String acceptedState = ack.getResultState() != null ? ack.getResultState()
+ : (decision == null && AnswerReconcileWork.isTerminal(ack.getTaskState()) ? ack.getTaskState() : state);
+ if (!AnswerReconcileWork.STATE_RUNNING.equals(acceptedState) && !AnswerReconcileWork.isTerminal(acceptedState)) {
+ throw new IllegalStateException("INVALID_RESULT_ACK：结果状态不合法");
+ }
+ work.setState(acceptedState);
+ work.setFinishedAt(AnswerReconcileWork.STATE_RUNNING.equals(acceptedState) ? null : LocalDateTime.now());
+ work.setNextRetryAt(null);
+ work.setLeaseUntil(AnswerReconcileWork.STATE_RUNNING.equals(acceptedState)
? LocalDateTime.now().plusSeconds(leaseSeconds) : null);
}
diff --git a/src/main/java/com/yunhai/svchall/service/impl/FlowApplyRecoveryService.java b/src/main/java/com/yunhai/svchall/service/impl/FlowApplyRecoveryService.java
index b49b3bd..d982010 100644
--- a/src/main/java/com/yunhai/svchall/service/impl/FlowApplyRecoveryService.java
+++ b/src/main/java/com/yunhai/svchall/service/impl/FlowApplyRecoveryService.java
@@ -97,8 +97,12 @@ public class FlowApplyRecoveryService {
* （重放时不再回读办件行——办件可能已被回调推进）。
*/
public FlowApplyCommand createInTransaction(ServiceApply apply, Integer intentId, String payloadJson) {
+ return createCommand(apply, intentId, payloadJson, 1);
+ }
+
+ private FlowApplyCommand createCommand(ServiceApply apply, Integer intentId, String payloadJson, int minimumAttempt) {
List history = commandRepository.findByApplyIdOrderByIdDesc(apply.getId());
- int attempt = history.isEmpty() ? 1 : maxAttempt(history) + 1;
+ int attempt = Math.max(minimumAttempt, history.isEmpty() ? 1 : maxAttempt(history) + 1);
FlowApplyCommand command = new FlowApplyCommand();
command.setApplyId(apply.getId());
command.setSchId(apply.getSchId());
@@ -118,6 +122,27 @@ public class FlowApplyRecoveryService {
return command;
}
+ /** 已授权用户入口复用同一命令通道；短事务领取/建命令后才出站，重复重提不叠加新轮。 */
+ public void requestExecution(Integer applyId, boolean resubmit) {
+ FlowApplyCommand selected = transactionTemplate.execute(tx -> {
+ ServiceApply apply = applyRepository.findByIdForUpdate(applyId)
+ .orElseThrow(() -> new BizException(ErrorCode.PARAM_ERROR, "办件不存在"));
+ List history = commandRepository.findByApplyIdOrderByIdDesc(applyId);
+ int localRound = FlowRoundContract.localRound(apply, history, objectMapper);
+ if (resubmit && !java.util.Arrays.asList(30, 40).contains(apply.getFlowStatus())) {
+ throw new BizException(ErrorCode.PARAM_ERROR, "仅驳回/已撤回状态的办件可重新提交");
+ }
+ FlowApplyCommand reusable = history.stream()
+ .filter(c -> !resubmit || (intOf(c.getAttempt()) > localRound
+ && !FlowApplyCommand.STATE_DONE.equals(c.getState()) && !FlowApplyCommand.STATE_DEAD.equals(c.getState())))
+ .max(java.util.Comparator.comparingInt(c -> intOf(c.getAttempt()))).orElse(null);
+ if (reusable != null) return reusable;
+ return createCommand(apply, apply.getFcIntentId(), payloadJson(apply), resubmit ? localRound + 1 : 1);
+ });
+ if (FlowApplyCommand.STATE_DONE.equals(selected.getState())) probe(selected);
+ else runCommand(selected.getId());
+ }
+
/** 事务提交后唤醒派发；无活动事务（单测/兜底）就地异步跑 */
public void dispatchAfterCommit(Integer commandId) {
if (TransactionSynchronizationManager.isSynchronizationActive()) {
@@ -142,7 +167,7 @@ public class FlowApplyRecoveryService {
}
}
- /** 周期恢复扫描（进程重启后追平；行锁串行，不与唤醒线程双发） */
+ /** 周期恢复扫描；领取短锁不覆盖 HTTP，重复出站由 FL token/轮次守卫吸收。 */
@Scheduled(fixedDelayString = "${svchall.flow-command.scan-interval-ms:30000}",
initialDelayString = "${svchall.flow-command.scan-initial-delay-ms:20000}")
public int recoverDue() {
@@ -184,20 +209,27 @@ public class FlowApplyRecoveryService {
return;
}
try {
- FlowClient.ApplyResult result = flowClient.applyWithIdentity(command.getSchId(), outbound);
transactionTemplate.executeWithoutResult(tx -> {
- if (command.getAttempt() != null && command.getAttempt() > 1) {
- applyRepository.markResubmitted(command.getApplyId(), result.getInstanceId(),
- LocalDateTime.now());
- } else {
- applyRepository.markApproving(command.getApplyId(), result.getInstanceId(),
- LocalDateTime.now());
- }
+ FlowApplyCommand row = commandRepository.findById(command.getId()).orElseThrow(IllegalStateException::new);
+ row.setSentAt(row.getSentAt() == null ? LocalDateTime.now() : row.getSentAt());
+ commandRepository.save(row);
});
- settle(command, FlowApplyCommand.STATE_DONE, null, result.getInstanceId(),
- ApplyFlowStatusEnum.APPROVING.getCode());
- log.info("[svc-hall] 命令已受理 commandId={} applyId={} instance={} duplicated={}",
- command.getId(), command.getApplyId(), result.getInstanceId(), result.isDuplicated());
+ FlowClient.ApplyResult result = flowClient.applyWithIdentity(command.getSchId(), outbound);
+ if (result == null || !FlowRoundContract.knownBranch(result.getBranch())) {
+ throw new IllegalStateException("FL_APPLY_CONTRACT：缺少或未知 branch，必须先查询真实状态");
+ }
+ int round = result.getApplyRound() == null ? 1 : result.getApplyRound();
+ if (result.getStatus() == null || result.getVersion() == null) {
+ Map remote = flowClient.getInstanceByBiz(command.getSchId(), command.getBizType(), command.getBizId());
+ completeFromQuery(command, remote, result.getInstanceId(), result.getBranch());
+ } else {
+ String remoteToken = round == intOf(command.getAttempt())
+ && ("CREATED".equals(result.getBranch()) || "RESTARTED".equals(result.getBranch())
+ || "SAME_REQUEST".equals(result.getBranch())) ? command.getRequestToken() : null;
+ completeRemote(command, result.getInstanceId(), result.getStatus(), result.getVersion(),
+ round, remoteToken, result.getBranch());
+ }
+
} catch (Exception e) {
// 任何异常都是"结果未知"：不许在这里判"应该没收到"，也不许在这里重发
markUnknown(command, e.toString());
@@ -225,28 +257,115 @@ public class FlowApplyRecoveryService {
}
if (remote == null) {
if (last != null) {
- bump(command, "查询远端状态失败，稍后再查：" + last);
+ bump(command, "查询远端状态失败，稍后再查：" + last, true);
return;
}
log.info("[svc-hall] 远端查无该业务键，补发同业务键同 token：commandId={} bizId={}",
command.getId(), command.getBizId());
+ outboundRetryOnly(command);
send(command);
return;
}
- Long instanceId = firstLong(remote.get("instanceId"), remote.get("id"));
- Integer status = firstInt(remote.get("status"));
- Integer version = firstInt(remote.get("version"));
- if (status == null) {
- bump(command, "远端返回无 status，继续查询");
- return;
+ try {
+ completeFromQuery(command, remote, null, null);
+ } catch (RuntimeException e) {
+ bump(command, "远端状态/轮次不可确认，继续查询：" + e.getMessage(), true);
}
- transactionTemplate.executeWithoutResult(tx -> applyRepository.mirrorRemoteState(
- command.getApplyId(), instanceId, status, version, LocalDateTime.now()));
- settle(command, FlowApplyCommand.STATE_DONE,
- "结果未知后查得远端 status=" + status + "（30/40 不重发，按 C.2 只回填）",
- instanceId, status);
}
+ private void outboundRetryOnly(FlowApplyCommand command) {
+ try {
+ com.fasterxml.jackson.databind.node.ObjectNode payload = (com.fasterxml.jackson.databind.node.ObjectNode) objectMapper.readTree(command.getPayloadJson());
+ payload.put("retryOnly", true);
+ command.setPayloadJson(objectMapper.writeValueAsString(payload));
+ transactionTemplate.executeWithoutResult(tx -> commandRepository.save(command));
+ } catch (Exception e) { throw new IllegalStateException("恢复载荷不可更新", e); }
+ }
+
+ private void completeFromQuery(FlowApplyCommand command, Map remote,
+ Long expectedInstance, String branch) {
+ if (remote == null) throw new IllegalStateException("FL_INSTANCE_CONTRACT：无实例");
+ Long id = firstLong(remote.get("instanceId"), remote.get("id"));
+ if (expectedInstance != null && !expectedInstance.equals(id)) {
+ throw new IllegalStateException("FL_INSTANCE_CONTRACT：apply 与查询实例不符");
+ }
+ Integer round = firstInt(remote.get("applyRound"));
+ String token = remote.get("requestToken") == null ? null : String.valueOf(remote.get("requestToken"));
+ boolean ownedRound = round != null && token != null && commandRepository.findByApplyIdOrderByIdDesc(command.getApplyId())
+ .stream().anyMatch(c -> round.equals(c.getAttempt()) && token.equals(c.getRequestToken())
+ && java.util.Objects.equals(c.getSchId(), command.getSchId())
+ && java.util.Objects.equals(c.getFlowGuid(), command.getFlowGuid())
+ && (c.getRemoteInstanceId() == null || c.getRemoteInstanceId().equals(id)));
+ if (!ownedRound) throw new IllegalStateException("FL_INSTANCE_CONTRACT：查询轮次/标识无法匹配本地命令");
+ completeRemote(command, id, firstInt(remote.get("status")), firstInt(remote.get("version")), round, token, branch);
+ }
+
+ private void completeRemote(FlowApplyCommand command, Long instanceId, Integer status, Integer version,
+ int remoteRound, String remoteToken, String branch) {
+ if (instanceId == null || instanceId < 1 || version == null || version < 0 || remoteRound < 1
+ || !java.util.Arrays.asList(10, 20, 30, 40).contains(status)) {
+ throw new IllegalStateException("FL_INSTANCE_CONTRACT：缺少合法实例/版本/状态/轮次");
+ }
+ if (("CREATED".equals(branch) || "RESTARTED".equals(branch)) && remoteRound != intOf(command.getAttempt())) {
+ throw new IllegalStateException("FL_APPLY_CONTRACT：新轮次与请求 attempt 不符");
+ }
+ transactionTemplate.executeWithoutResult(tx -> {
+ FlowApplyCommand row = commandRepository.findByIdForUpdate(command.getId()).orElseThrow(IllegalStateException::new);
+ int confirmedRound = FlowRoundContract.outcome(row, objectMapper).path("applyRound").asInt(0);
+ if (confirmedRound > remoteRound || (confirmedRound == remoteRound && row.getRemoteVersion() != null
+ && row.getRemoteVersion() > version)) return; // 较旧 HTTP 响应不能倒写命令台账。
+ if (confirmedRound == remoteRound && version.equals(row.getRemoteVersion()) && row.getRemoteStatus() != null
+ && ApplyFlowStatusEnum.byCode(row.getRemoteStatus()).terminal() && !status.equals(row.getRemoteStatus())) return;
+ ServiceApply apply = applyRepository.findByIdForUpdate(command.getApplyId()).orElseThrow(IllegalStateException::new);
+ if (apply.getFlowInstanceId() != null && !apply.getFlowInstanceId().equals(instanceId)) {
+ throw new IllegalStateException("FL_INSTANCE_CONTRACT：实例归属与本地绑定不符");
+ }
+ int localRound = FlowRoundContract.localRound(apply,
+ commandRepository.findByApplyIdOrderByIdDesc(command.getApplyId()), objectMapper);
+ int localVersion = apply.getLastInstanceVersion() == null ? 0 : apply.getLastInstanceVersion();
+ boolean provedNewRound = remoteRound > localRound && remoteRound == intOf(command.getAttempt())
+ && command.getRequestToken().equals(remoteToken)
+ && (branch == null || "CREATED".equals(branch) || "RESTARTED".equals(branch) || "SAME_REQUEST".equals(branch));
+ boolean reopen = ApplyFlowStatusEnum.byCode(apply.getFlowStatus()).terminal() && status == 10;
+ boolean sameVersionConflict = version == localVersion
+ && ApplyFlowStatusEnum.byCode(apply.getFlowStatus()).terminal() && !status.equals(apply.getFlowStatus());
+ if (remoteRound >= localRound && version >= localVersion && !sameVersionConflict
+ && (!reopen || (provedNewRound && version > localVersion))) {
+ boolean advanceRound = remoteRound > localRound;
+ boolean sameVersionTerminal = version == localVersion && ApplyFlowStatusEnum.byCode(apply.getFlowStatus()).terminal();
+ apply.setFlowInstanceId(instanceId);
+ apply.setFlowStatus(status);
+ apply.setLastInstanceVersion(version);
+ apply.setSubmitTime(apply.getSubmitTime() == null || advanceRound ? LocalDateTime.now() : apply.getSubmitTime());
+ apply.setFinishTime(status == 10 ? null : (apply.getFinishTime() == null || advanceRound ? LocalDateTime.now() : apply.getFinishTime()));
+ try {
+ com.fasterxml.jackson.databind.node.ObjectNode result = !advanceRound || sameVersionTerminal
+ ? (com.fasterxml.jackson.databind.node.ObjectNode) objectMapper.readTree(
+ apply.getFlowResultJson() == null ? "{}" : apply.getFlowResultJson())
+ : objectMapper.createObjectNode();
+ result.put("applyRound", remoteRound);
+ apply.setFlowResultJson(objectMapper.writeValueAsString(result));
+ } catch (Exception e) { throw new IllegalStateException("本地轮次证据不可写入", e); }
+ applyRepository.save(apply);
+ }
+ try {
+ com.fasterxml.jackson.databind.node.ObjectNode payload = (com.fasterxml.jackson.databind.node.ObjectNode)
+ objectMapper.readTree(row.getPayloadJson());
+ com.fasterxml.jackson.databind.node.ObjectNode outcome = payload.putObject("_flOutcome");
+ outcome.put("branch", branch); outcome.put("applyRound", remoteRound);
+ outcome.put("status", status); outcome.put("version", version); outcome.put("instanceId", instanceId);
+ outcome.put("requestToken", remoteToken);
+ row.setPayloadJson(objectMapper.writeValueAsString(payload));
+ } catch (Exception e) { throw new IllegalStateException("远端分支证据不可写入", e); }
+ row.setRemoteInstanceId(instanceId); row.setRemoteStatus(status); row.setRemoteVersion(version);
+ row.setState(FlowApplyCommand.STATE_DONE); row.setSettledAt(LocalDateTime.now());
+ row.setNextRetryAt(null); row.setLastError(null);
+ commandRepository.save(row);
+ });
+ }
+
+ private static int intOf(Integer value) { return value == null ? 0 : value; }
+
// ==================== 状态落定 ====================
private void settle(FlowApplyCommand command, String state, String note) {
@@ -291,6 +410,10 @@ public class FlowApplyRecoveryService {
/** 退避重试（含"未接入"等待）；耗尽即 DEAD，留可查记录，不无限循环 */
private void bump(FlowApplyCommand command, String error) {
+ bump(command, error, false);
+ }
+
+ private void bump(FlowApplyCommand command, String error, boolean remainUnknown) {
transactionTemplate.executeWithoutResult(tx -> {
FlowApplyCommand row = commandRepository.findById(command.getId()).orElse(null);
if (row == null) {
@@ -305,7 +428,7 @@ public class FlowApplyRecoveryService {
log.error("[svc-hall] 出站命令重试耗尽→待人工处置：commandId={} applyId={} retries={} last={}",
row.getId(), row.getApplyId(), retries, error);
} else {
- row.setState(FlowApplyCommand.STATE_FAILED);
+ row.setState(remainUnknown ? FlowApplyCommand.STATE_UNKNOWN : FlowApplyCommand.STATE_FAILED);
row.setNextRetryAt(LocalDateTime.now().plusSeconds(backoffSeconds(retries)));
}
commandRepository.save(row);
diff --git a/src/main/java/com/yunhai/svchall/service/impl/FlowCenterClient.java b/src/main/java/com/yunhai/svchall/service/impl/FlowCenterClient.java
index 2a9b671..28963fa 100644
--- a/src/main/java/com/yunhai/svchall/service/impl/FlowCenterClient.java
+++ b/src/main/java/com/yunhai/svchall/service/impl/FlowCenterClient.java
@@ -161,7 +161,17 @@ public class FlowCenterClient implements FlowClient {
throw new BizException(ErrorCode.PARAM_ERROR, "流程中心代发起返回异常（缺少 instanceId）");
}
return new ApplyResult(data.get("instanceId").asLong(),
- data.has("duplicated") && data.get("duplicated").asBoolean());
+ data.path("duplicated").asBoolean(), data.path("branch").asText(null),
+ v2Integer(data, "applyRound"), v2Integer(data, "status"), v2Integer(data, "version"));
+ }
+
+ private static Integer v2Integer(JsonNode data, String key) {
+ JsonNode value = data.get(key);
+ if (value == null || value.isNull()) return null;
+ if (!value.isIntegralNumber() || !value.canConvertToInt()) {
+ throw new BizException(ErrorCode.PARAM_ERROR, "FL_APPLY_CONTRACT：" + key + " 必须为整数");
+ }
+ return value.intValue();
}
@Override
diff --git a/src/main/java/com/yunhai/svchall/service/impl/FlowRoundContract.java b/src/main/java/com/yunhai/svchall/service/impl/FlowRoundContract.java
new file mode 100644
index 0000000..e620d9d
--- /dev/null
+++ b/src/main/java/com/yunhai/svchall/service/impl/FlowRoundContract.java
@@ -0,0 +1,35 @@
+package com.yunhai.svchall.service.impl;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.yunhai.svchall.entity.FlowApplyCommand;
+import com.yunhai.svchall.entity.ServiceApply;
+import java.util.Arrays;
+import java.util.List;
+
+/** 同一处解析持久轮次证据；不把本地请求 attempt 当成远端已接受轮次。 */
+final class FlowRoundContract {
+ private FlowRoundContract() { }
+
+ static boolean knownBranch(String branch) {
+ return Arrays.asList("CREATED", "IDEMPOTENT", "SAME_REQUEST", "STALE_ROUND", "REPLAY_ONLY",
+ "RESTARTED", "CONCURRENT_CREATED").contains(branch);
+ }
+
+ static JsonNode outcome(FlowApplyCommand command, ObjectMapper mapper) {
+ try { return mapper.readTree(command.getPayloadJson()).path("_flOutcome"); }
+ catch (Exception e) { return mapper.createObjectNode(); }
+ }
+
+ static int localRound(ServiceApply apply, List history, ObjectMapper mapper) {
+ int round = 1;
+ try { round = Math.max(round, mapper.readTree(apply.getFlowResultJson()).path("applyRound").asInt(1)); }
+ catch (Exception ignored) { }
+ for (FlowApplyCommand command : history) {
+ if (apply.getFlowInstanceId() != null && apply.getFlowInstanceId().equals(command.getRemoteInstanceId())) {
+ round = Math.max(round, outcome(command, mapper).path("applyRound").asInt(1));
+ }
+ }
+ return round;
+ }
+}
diff --git a/src/main/java/com/yunhai/svchall/service/impl/ServiceApplyService.java b/src/main/java/com/yunhai/svchall/service/impl/ServiceApplyService.java
index 81d9093..11e11ad 100644
--- a/src/main/java/com/yunhai/svchall/service/impl/ServiceApplyService.java
+++ b/src/main/java/com/yunhai/svchall/service/impl/ServiceApplyService.java
@@ -572,6 +572,13 @@ public class ServiceApplyService {
}
ServiceApply apply = applyRepository.findById(applyId)
.orElseThrow(() -> new BizException(ErrorCode.PARAM_ERROR, "办件不存在"));
+ if (flowCommands != null) {
+ flowCommands.requestExecution(applyId, resubmit);
+ return; // 生产装配一律消费 v2 分支，不能回落到旧 apply + 常量 10。
+ }
+ if (apply.getFcIntentId() != null || apply.getFcAnswerUuid() != null) {
+ throw new BizException(ErrorCode.INTERNAL_SERVER_ERROR, "可信办件命令服务未装配，拒绝走旧发起路径");
+ }
try {
FlowClient.ApplyResult result = flowClient.apply(
apply.getSchId(),
diff --git a/src/main/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandler.java b/src/main/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandler.java
index 53a88ce..5bfd905 100644
--- a/src/main/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandler.java
+++ b/src/main/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandler.java
@@ -40,6 +40,9 @@ public class SvcFlowCallbackHandler implements FlowCallbackHandler {
private final ServiceApplyRepository applyRepository;
private final ObjectMapper objectMapper;
+ @org.springframework.beans.factory.annotation.Autowired
+ private com.yunhai.svchall.repository.FlowApplyCommandRepository commandRepository;
+
@Override
public void onEvent(FlowCallbackEvent event) {
Integer applyId;
@@ -81,26 +84,24 @@ public class SvcFlowCallbackHandler implements FlowCallbackHandler {
event.getInstanceId(), version, apply.getLastInstanceVersion());
return;
}
- advance(apply, event.getEvent(), version, event);
+ Integer round = verifiedRound(apply, event);
+ if (round == null) return;
+ int previousRound = currentRound(apply);
+ if (round < previousRound) return;
+ advance(apply, event.getEvent(), version, event, round, round > previousRound);
log.info("[svc-hall] 办件镜像状态推进 applyId={} instanceId={} event={} → status={}",
applyId, event.getInstanceId(), event.getEvent(), apply.getFlowStatus());
}
- /**
- * 状态推进（幂等：同一事件重放由 eventId 去重拦截，此处只需正确覆盖目标态）。
- * 终态守卫按版本感知（P1-03）：行已终态时，仅当事件版本>已消费版本（=新一轮
- * R4 重启的回调早于 markResubmitted 回开的竞态窗口）才放行推进——新轮回调先到/新轮
- * 快速终态两类时序不再被"旧结果非空"一刀切拒绝；旧轮迟到/重复（版本不高于已消费）
- * 仍按原守卫拒收，不回退不覆盖。
- */
- private void advance(ServiceApply apply, String event, Integer version, FlowCallbackEvent payload) {
+ /** 只有带匹配命令 attempt/token 的更高轮次才能回开终态；实例版本只负责轮内乱序。 */
+ private void advance(ServiceApply apply, String event, Integer version, FlowCallbackEvent payload, int round, boolean provedNewRound) {
Integer status = mapStatus(event);
if (status == null) {
log.warn("[svc-hall] 未知回调事件类型：{}（忽略）", event);
return;
}
if (ApplyFlowStatusEnum.byCode(apply.getFlowStatus()).terminal()) {
- boolean newRound = version != null && apply.getLastInstanceVersion() != null
+ boolean newRound = provedNewRound && version != null && apply.getLastInstanceVersion() != null
&& version > apply.getLastInstanceVersion();
if (!newRound) {
// P1-3 终态不回退守卫：终态事件与最后一个 task.handled 同 instanceVersion，若 handled
@@ -124,6 +125,7 @@ public class SvcFlowCallbackHandler implements FlowCallbackHandler {
}
Map result = new HashMap<>();
result.put("event", event);
+ result.put("applyRound", round);
Map data = payload.getData();
if (data != null) {
copyIfPresent(data, result, "result");
@@ -150,6 +152,38 @@ public class SvcFlowCallbackHandler implements FlowCallbackHandler {
applyRepository.save(apply);
}
+ private int currentRound(ServiceApply apply) {
+ java.util.List history = commandRepository == null
+ ? java.util.Collections.emptyList() : commandRepository.findByApplyIdOrderByIdDesc(apply.getId());
+ return FlowRoundContract.localRound(apply, history, objectMapper);
+ }
+
+ private Integer verifiedRound(ServiceApply apply, FlowCallbackEvent event) {
+ Map data = event.getData();
+ Object rawRound = data == null ? null : data.get("applyRound");
+ String token = data == null || data.get("requestToken") == null ? null : String.valueOf(data.get("requestToken"));
+ java.util.List history = commandRepository == null
+ ? java.util.Collections.emptyList() : commandRepository.findByApplyIdOrderByIdDesc(apply.getId());
+ if (rawRound == null && token == null && history.isEmpty()) return 1; // 真正的无命令存量单；不据版本回开终态。
+ int round;
+ try { round = Integer.parseInt(String.valueOf(rawRound)); }
+ catch (RuntimeException e) { return null; }
+ if (round < 1 || token == null || commandRepository == null) return null;
+ com.yunhai.svchall.entity.FlowApplyCommand command = commandRepository
+ .findByApplyIdAndAttempt(apply.getId(), round).orElse(null);
+ if (command == null || command.getSentAt() == null || !token.equals(command.getRequestToken())
+ || !java.util.Objects.equals(command.getSchId(), apply.getSchId())
+ || !java.util.Objects.equals(command.getFlowGuid(), event.getFlowGuid())
+ || (apply.getFlowGuid() != null && !apply.getFlowGuid().equals(event.getFlowGuid()))
+ || (command.getRemoteInstanceId() != null && !command.getRemoteInstanceId().equals(event.getInstanceId()))) {
+ log.warn("[svc-hall] 回调轮次/命令/实例不匹配，拒收 applyId={} round={}", apply.getId(), round);
+ return null;
+ }
+ com.fasterxml.jackson.databind.JsonNode outcome = FlowRoundContract.outcome(command, objectMapper);
+ if (outcome.has("applyRound") && outcome.path("applyRound").asInt() != round) return null;
+ return round;
+ }
+
private static void copyIfPresent(Map data, Map target, String key) {
Object v = data.get(key);
if (v != null) {
diff --git a/src/test/java/com/yunhai/svchall/service/impl/FlowCenterClientFacadeTest.java b/src/test/java/com/yunhai/svchall/service/impl/FlowCenterClientFacadeTest.java
index 563c43a..d1d1ec5 100644
--- a/src/test/java/com/yunhai/svchall/service/impl/FlowCenterClientFacadeTest.java
+++ b/src/test/java/com/yunhai/svchall/service/impl/FlowCenterClientFacadeTest.java
@@ -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));
+ }
+
}
diff --git a/src/test/java/com/yunhai/svchall/service/impl/ServiceApplyFlowSubmitTest.java b/src/test/java/com/yunhai/svchall/service/impl/ServiceApplyFlowSubmitTest.java
index 5a8660f..72170fb 100644
--- a/src/test/java/com/yunhai/svchall/service/impl/ServiceApplyFlowSubmitTest.java
+++ b/src/test/java/com/yunhai/svchall/service/impl/ServiceApplyFlowSubmitTest.java
@@ -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());
+ }
+
}
diff --git a/src/test/java/com/yunhai/svchall/service/impl/ShAnswerFlowTest.java b/src/test/java/com/yunhai/svchall/service/impl/ShAnswerFlowTest.java
index bff2c53..bad2240 100644
--- a/src/test/java/com/yunhai/svchall/service/impl/ShAnswerFlowTest.java
+++ b/src/test/java/com/yunhai/svchall/service/impl/ShAnswerFlowTest.java
@@ -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 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 created = new ArrayList<>();
diff --git a/src/test/java/com/yunhai/svchall/service/impl/ShReconcileTaskTest.java b/src/test/java/com/yunhai/svchall/service/impl/ShReconcileTaskTest.java
index c38d359..7b589dd 100644
--- a/src/test/java/com/yunhai/svchall/service/impl/ShReconcileTaskTest.java
+++ b/src/test/java/com/yunhai/svchall/service/impl/ShReconcileTaskTest.java
@@ -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;
diff --git a/src/test/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandlerTest.java b/src/test/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandlerTest.java
index ee2d232..7514198 100644
--- a/src/test/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandlerTest.java
+++ b/src/test/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandlerTest.java
@@ -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 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());
+ }
+
}