Author SHA1 Message Date
chenxing a5af47abfa Run mandatory RC gates with isolated CI failure checks
build / build-push (push) Successful in 3m8s
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.
2026-10-23 10:51:42 +08:00
chenxing a4d220362e Preserve active leases and align flow recovery rounds
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.
2026-10-23 10:51:40 +08:00
chenxing 8edccd5514 Preserve reviewed existing gate prerequisites
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.
2026-10-23 10:51:32 +08:00
21 changed files with 1111 additions and 87 deletions
+25 -2
View File
@@ -9,13 +9,17 @@ name: build
on:
push:
branches: [ main ]
branches: [ main, codex/core-rc-20261023 ]
workflow_dispatch:
inputs:
cache_bust:
description: '构建前清空 maven m2 缓存'
required: false
default: 'false'
tests_only:
description: '只验证测试和 JAR,不构建推送镜像或通知建档'
required: false
default: 'true'
jobs:
build-push:
@@ -36,6 +40,10 @@ jobs:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
git clone --depth 1 --branch main "http://x-access-token:${GITHUB_TOKEN}@gitea.yunhaixingtong.com/${{ github.repository }}.git" /tmp/src
git -C /tmp/src fetch --depth 1 origin "$GITHUB_SHA"
git -C /tmp/src checkout --detach FETCH_HEAD
test "$(git -C /tmp/src rev-parse HEAD)" = "$GITHUB_SHA"
echo "CI_SOURCE_SHA=$GITHUB_SHA"
cp -a /tmp/src/. .
rm -rf /tmp/src
# 取提交备注首行,脱去换行/双引号/反斜杠(避免破坏 webhook JSON),截断到 200 字符
@@ -64,6 +72,19 @@ jobs:
# GITHUB_RUN_ID 实例级唯一,再拼时间戳兜底。
BUILD_CONTAINER="mavenbuild-${GITHUB_RUN_ID:-0}-$(date +%s)"
echo "BUILD_CONTAINER=${BUILD_CONTAINER}"
cleanup_ci() {
result=$?
trap - EXIT
docker rm -fv "$BUILD_CONTAINER" >/dev/null 2>&1 || true
docker info >/dev/null 2>&1 || { echo 'CI_CONTAINER_CLEANUP=UNVERIFIED' >&2; exit 1; }
if docker inspect "$BUILD_CONTAINER" >/dev/null 2>&1; then
echo 'CI_CONTAINER_CLEANUP=FAILED' >&2
exit 1
fi
echo 'CI_CONTAINER_CLEANUP=PASS'
exit "$result"
}
trap cleanup_ci EXIT
docker rm -f "$BUILD_CONTAINER" 2>/dev/null || true
docker run -d --name "$BUILD_CONTAINER" \
--network infra-infra-net \
@@ -71,13 +92,14 @@ jobs:
-w /build maven:3.6.3-jdk-8 sleep 3600
docker cp "${PWD}/." "$BUILD_CONTAINER":/build/
docker exec "$BUILD_CONTAINER" sh -c "printf '%s' '<settings><profiles><profile><id>nexus-repo</id><repositories><repository><id>nexus-internal</id><url>http://nexus:8081/repository/maven-public/</url><releases><enabled>true</enabled></releases><snapshots><enabled>true</enabled><updatePolicy>always</updatePolicy></snapshots></repository></repositories></profile></profiles><activeProfiles><activeProfile>nexus-repo</activeProfile></activeProfiles></settings>' > /root/.m2/settings.xml"
docker exec "$BUILD_CONTAINER" sh -c 'cd /build && mvn -B -DskipTests clean package'
docker exec "$BUILD_CONTAINER" sh -c 'cd /build && sh ci/test-gate.sh clean package'
docker exec "$BUILD_CONTAINER" sh -c 'cd /build/target && cp *.jar app.jar'
mkdir -p "${PWD}/target"
docker cp "$BUILD_CONTAINER":/build/target/app.jar "${PWD}/target/app.jar"
docker rm -f "$BUILD_CONTAINER"
- name: 构建并推送镜像
if: ${{ success() && github.ref == 'refs/heads/main' && inputs.tests_only != 'true' }}
run: |
set -e
TAG=$(echo "$GITHUB_SHA" | cut -c1-8)
@@ -88,6 +110,7 @@ jobs:
# 通知 projman:webhook 收到后只建档 Release(status=pending),部署由运维在 DeployTab 手动触发(App.deployMode=manual)
- name: 通知 projman 建档(部署由运维手动触发)
if: ${{ success() && github.ref == 'refs/heads/main' && inputs.tests_only != 'true' }}
run: |
curl -sS -X POST "http://192.168.10.160:8090/projman/api/deploy/webhook" \
-H "Content-Type: application/json" \
+13
View File
@@ -0,0 +1,13 @@
#!/bin/sh
set -eu
cd "$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)"
TESTS=ShAnswerFlowTest,ShReconcileExtensionTest,ShReconcileTaskTest,ShMqEntryDefaultOffTest,ShProjectionRepairSqlGuardTest,ShFormBindingTest,SvcFlowCallbackHandlerTest,FlowReconcileJobTest,ServiceApplyFlowSubmitTest,FlowCenterClientFacadeTest,FlowCenterClientRoutingTest
# Surefire 的选择器部分匹配也可能成功;每个指定类必须随源码存在。
oldIFS=$IFS
IFS=,
for test in $TESTS; do
[ -n "$(find src/test/java -type f -name "$test.java" -print -quit)" ] || { echo "Missing required test class: $test" >&2; exit 2; }
done
IFS=$oldIFS
if [ "$#" -eq 0 ]; then set -- test; fi
exec mvn -B "$@" -Dtest="$TESTS" -DfailIfNoTests=true -Dsurefire.failIfNoSpecifiedTests=true -DforkCount=0 -Dspring.jpa.hibernate.ddl-auto=validate < /dev/null
+53
View File
@@ -0,0 +1,53 @@
# SH 流程恢复与回调轮次契约(2026-10-23)
复用既有 flow_apply_command,不新增库列或替代流程架构。FC 签名收据 + HTTP 建件仍是主链,MQ 不作为必选。
## 请求轮次与实际轮次
command.attempt 记录用户已授权请求的历史,不根据 duplicated 猜测,也不把远端 applyRound 写回 attempt。
FL 实际结果(branch/applyRound/status/version/instanceId)存于原 payload_json 的服务器保留节 `_flOutcome`,
原发起材料字段不变;再次发送反序列化 FlowOutbound 时不把该保留节送出。
service_apply.flow_result_json 的 applyRound 是已确认镜像轮次;没有字段的旧行初始按 1。
| FL 原分支 | SH 行为 |
|---|---|
| CREATED | 按真实状态/版本确认新建;不假定 START 后必然仍为 10 |
| IDEMPOTENT | 镜像真实 10/20;本地终态不凭幂等响应回开 |
| SAME_REQUEST | 同命令回读,不新建 attempt;30/40 仍为原终态。若是已重启命令的丢响应恢复,只凭同 token+匹配命令轮次修正已存在的远端轮次 |
| STALE_ROUND | 保存远端实际轮次与状态;不倒退更高的本地镜像轮次、不回开终态 |
| REPLAY_ONLY | 保留远端真实状态;网络恢复不得借此重启 |
| RESTARTED | 唯一主动进入新轮的 FL 分支;回填还必须通过轮次、token、实例与版本保护 |
| CONCURRENT_CREATED | 当作已存在实例回读,按实际状态确认,不依据本地 attempt 开新轮 |
未知/缺失 branch 不是成功,转 UNKNOWN 先查真实状态。旧 FL 响应缺 status/version 时同样查询,
查询结果必须用 instanceId 与本地命令的 applyRound/requestToken/学校/流程线交叉确认。
UNKNOWN 查询异常或证据不符继续 UNKNOWN + 退避,不能降为下一次会盲发的 FAILED;预算耗尽 DEAD。
查无才同 token 补发并置 retryOnly=true;只有用户明确重提才可以建立新的请求轮。
## 用户入口与并发
ServiceApplyService 在生产装配 flowCommands 时,初次代发起、retry、resubmit 全走同一命令服务;
可信 FC 办件缺命令服务直接拒绝,不能绕到旧 apply + markResubmitted 常量 10。
用户授权重提在短事务中锁办件,已有尚未落定的新轮命令时复用它,双击不再创建第三轮;
真正开始重提的命令 attempt 至少是当前镜像轮次+1。HTTP 不占办件长事务。
无 Spring 命令装配的非 FC 历史适配路径保留,不能用于可信 FC 办件。
领取短锁不覆盖 HTTP,重复请求可能并发;FL 既有 token/轮次守卫吸收重复,
SH 结算时锁命令并检查轮次/版本,旧响应既不能覆盖业务终态,也不能倒写较新命令台账。
## 回调
FL 在同步事件发布阶段冻结 applyRound/requestToken,AFTER_COMMIT 不用重读到的新轮为旧事件贴标。
两个字段在既有回调 data 中传递,无需修改共享 SDK DTO 或数据库。
SH 必须同时匹配事件实例、学校、流程线,以及该办件对应的 command.attempt/requestToken;
命令必须已发送,若已有 FL 结果证据,其实际 applyRound 必须一致。
只有经过这些校验的更高轮次 + 更高实例版本才可以从终态回开;实例版本只承担轮内乱序判据。
旧轮即使版本很高,也不能修改新轮终态或结果;同轮终态也不被迟到 task.* 回退。
新轮快速结束的回调先到后,较旧的 apply 响应不覆盖它。
缺轮次身份的有命令回调保守忽略;真正无命令存量单只走第 1 轮兼容,绝不单凭版本回开终态。
## 验证边界
规则测试控制竞争时序,不声称为 MySQL 并发或 160 验收。FL 指定 11 项仍为真实 MySQL 测试;
CI 只连接本次 job 的独立 schema,Hibernate 固定 validate。部署必须先 FC/FL 后 SH,
以便 SH 严格校验启用前,FL 已能发送冻结轮次身份;存量缺身份积压需单独清点,不回退新数据安全语义。
@@ -193,6 +193,9 @@ public final class FormSyncModels {
private Boolean duplicate;
private Boolean staleAttempt;
private String taskState;
private String decision;
private Integer reportAttempt;
private String resultState;
private String message;
}
@@ -21,6 +21,7 @@ import java.util.Optional;
*/
public interface AnswerReconcileWorkRepository extends JpaRepository<AnswerReconcileWork, Integer> {
@Lock(LockModeType.PESSIMISTIC_WRITE)
Optional<AnswerReconcileWork> findByTaskUuid(String taskUuid);
@Lock(LockModeType.PESSIMISTIC_WRITE)
@@ -15,9 +15,8 @@ import java.util.Optional;
/**
* 出站发起命令仓库(TASK_2 C.1/C.2)。
*
* <p>{@link #findByIdForUpdate} 与既有 {@code ServiceApplyRepository#findByIdForUpdate} 同构:
* 调度线程池与恢复扫描可能同时盯上同一条命令,行锁把两者串起来,避免"两个线程各自以为
* 自己是唯一发送者"而双发(FL 侧 uk 与 requestToken 是第二、第三道闸)。</p>
* <p>短事务行锁用于读取和结算持久命令,不跨 HTTP 持有;同命令并发发送仍可能发生。
* 远端按 requestToken/attempt 幂等吸收,本地结算再锁命令并按远端轮次/版本防止旧响应倒写。</p>
*/
public interface FlowApplyCommandRepository extends JpaRepository<FlowApplyCommand, Integer> {
@@ -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; }
}
}
@@ -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);
}
@@ -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<FlowApplyCommand> 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<FlowApplyCommand> 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<String, Object> 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<String, Object> 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);
@@ -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
@@ -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<FlowApplyCommand> 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;
}
}
@@ -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(),
@@ -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 去重拦截,此处只需正确覆盖目标态)。
* 终态守卫按<b>版本感知</b>(P1-03):行已终态时,仅当事件版本&gt;已消费版本(=新一轮
* 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<String, Object> result = new HashMap<>();
result.put("event", event);
result.put("applyRound", round);
Map<String, Object> 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<com.yunhai.svchall.entity.FlowApplyCommand> 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<String, Object> 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<com.yunhai.svchall.entity.FlowApplyCommand> 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<String, Object> data, Map<String, Object> target, String key) {
Object v = data.get(key);
if (v != null) {
@@ -42,6 +42,16 @@ class FlowReconcileJobTest {
flowClient = mock(FlowClient.class);
}
/** C.1 起 Job 依赖出站命令仓判"新命令链办件归恢复服务管";测试注入空历史桩,否则 NPE */
private FlowReconcileJob job(boolean enabled) {
FlowReconcileJob job = new FlowReconcileJob(applyRepository, flowClient, enabled);
com.yunhai.svchall.repository.FlowApplyCommandRepository commands =
mock(com.yunhai.svchall.repository.FlowApplyCommandRepository.class);
when(commands.findByApplyIdOrderByIdDesc(anyInt())).thenReturn(Collections.emptyList());
ReflectionTestUtils.setField(job, "commandRepository", commands);
return job;
}
private ServiceApply apply(int status, Integer lastVersion) {
ServiceApply apply = new ServiceApply();
ReflectionTestUtils.setField(apply, "id", 42);
@@ -54,7 +64,7 @@ class FlowReconcileJobTest {
@Test
void disabledByDefaultJobDoesNothing() {
FlowReconcileJob job = new FlowReconcileJob(applyRepository, flowClient, false);
FlowReconcileJob job = job(false);
job.reconcile();
verify(flowClient, never()).enabled();
verify(applyRepository, never()).findByFlowStatus(anyInt(), any(Pageable.class));
@@ -71,7 +81,7 @@ class FlowReconcileJobTest {
remote.put("version", 4);
when(flowClient.getInstanceByBiz(1L, "svc_item", "42")).thenReturn(remote);
new FlowReconcileJob(applyRepository, flowClient, true).reconcile();
job(true).reconcile();
assertEquals(ApplyFlowStatusEnum.COMPLETED.getCode(), apply.getFlowStatus());
assertEquals(Integer.valueOf(4), apply.getLastInstanceVersion());
@@ -90,7 +100,7 @@ class FlowReconcileJobTest {
remote.put("version", 2);
when(flowClient.getInstanceByBiz(anyLong(), anyString(), anyString())).thenReturn(remote);
new FlowReconcileJob(applyRepository, flowClient, true).reconcile();
job(true).reconcile();
verify(applyRepository, never()).save(any());
}
@@ -98,7 +108,7 @@ class FlowReconcileJobTest {
@Test
void enabledJobSkipsWhenFlowCenterDisabled() {
when(flowClient.enabled()).thenReturn(false);
new FlowReconcileJob(applyRepository, flowClient, true).reconcile();
job(true).reconcile();
verify(applyRepository, never()).findByFlowStatus(anyInt(), any(Pageable.class));
}
@@ -111,7 +121,7 @@ class FlowReconcileJobTest {
when(flowClient.getInstanceByBiz(anyLong(), anyString(), anyString()))
.thenThrow(new RuntimeException("boom"));
new FlowReconcileJob(applyRepository, flowClient, true).reconcile();
job(true).reconcile();
verify(applyRepository, never()).save(any());
}
@@ -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));
}
}
@@ -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());
}
}
@@ -34,6 +34,7 @@ import com.yunhai.svchall.service.SiteDirectory;
import com.yunhai.svchall.service.SiteGuard;
import com.yunhai.svchall.service.impl.AnswerProjectionStore.Disposition;
import com.yunhai.svchall.service.impl.AnswerProjectionStore.Outcome;
import com.yunhai.svchall.service.impl.AnswerProjectionStore.RepairResult;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
@@ -320,6 +321,70 @@ class ShAnswerFlowTest {
assertThat(h.projectionsSaved.get(0).getLinkState()).isEqualTo(FcAnswerProjection.LINK_APPLIED);
}
@Test
@DisplayName("已受理后答卷再演进:普通投影跟随最新版,意图受理坐标与办件关联保持固化(决策③)")
void appliedEvidenceStaysFrozenWhenNewerRevisionArrives() {
Harness h = harness();
FcApplyIntent intent = intent(12, 9001);
intent.setState(FcApplyIntent.STATE_APPLIED);
intent.setFcIntentId(FC_INTENT_ID);
intent.setAnswerUuid(ANSWER_UUID);
intent.setAnswerRevision(1L);
String acceptedHash = hash(snapshot("{\"k\":\"v1\"}"));
intent.setContentHash(acceptedHash);
h.intentRows.add(intent);
FcAnswerProjection applied = projection(1L, acceptedHash);
applied.setValuesJson("{\"k\":\"v1\"}");
applied.setIntentId(12);
applied.setApplyId(9001);
applied.setLinkState(FcAnswerProjection.LINK_APPLIED);
h.currentProjection = applied;
SnapshotEnvelope newer = envelope(2L, 7, snapshot("{\"k\":\"v2\"}"));
newer.getSnapshot().setIntent(mapper.createObjectNode().put("intentId", FC_INTENT_ID));
Disposition result = h.store.applySnapshot(newer);
assertThat(result.getApplyId()).isEqualTo(9001);
assertThat(h.applyCounter.get()).isZero();
// 普通投影(数据面镜像)跟随最新答卷
assertThat(applied.getRevision()).isEqualTo(2L);
assertThat(applied.getValuesJson()).isEqualTo("{\"k\":\"v2\"}");
// 审批证据锚点:意图的受理坐标只记受理那一刻,不被后来事件改写
assertThat(intent.getAnswerRevision()).isEqualTo(1L);
assertThat(intent.getContentHash()).isEqualTo(acceptedHash);
assertThat(applied.getLinkState()).isEqualTo(FcAnswerProjection.LINK_APPLIED);
assertThat(applied.getIntentId()).isEqualTo(12);
assertThat(applied.getApplyId()).isEqualTo(9001);
}
@Test
@DisplayName("权威修复打在已建件行上:只刷快照镜像,关联列不在条件更新里,也不走建件(决策③)")
void authoritativeRepairOfAppliedRowMirrorsOnly() {
Harness h = harness();
FcAnswerProjection applied = projection(1L, hash(snapshot("{\"k\":\"v1\"}")));
applied.setValuesJson("{\"k\":\"v1\"}");
applied.setIntentId(12);
applied.setApplyId(9001);
applied.setLinkState(FcAnswerProjection.LINK_APPLIED);
h.currentProjection = applied;
when(h.projections.repairIfUnchanged(any(), any(), any(), any())).thenReturn(1);
RepairResult result = h.store.repairAuthoritative(envelope(5L, 7, snapshot("{\"k\":\"v5\"}")),
1L, hash(snapshot("{\"k\":\"v1\"}")), 0, "proof-1");
assertThat(result).isEqualTo(RepairResult.APPLIED);
ArgumentCaptor<FcAnswerProjection> replacement = ArgumentCaptor.forClass(FcAnswerProjection.class);
verify(h.projections).repairIfUnchanged(replacement.capture(), eq(1L),
eq(hash(snapshot("{\"k\":\"v1\"}"))), eq(0));
// 关联列不在 UPDATE SET 里(数据库语义由 ShProjectionRepairSqlGuardTest 钉住),replacement 里根本没有它们
assertThat(replacement.getValue().getIntentId()).isNull();
assertThat(replacement.getValue().getApplyId()).isNull();
assertThat(replacement.getValue().getValuesJson()).isEqualTo("{\"k\":\"v5\"}");
// 修复不产生任何业务副作用:不建件、不回写办件证据
verify(h.projections, never()).save(any());
verify(h.applyService, never()).createFromVerifiedIntent(any(), any());
}
// ==================== 意图入口与确认(A.1/A.2/A.4) ====================
@Test
@@ -526,6 +591,136 @@ class ShAnswerFlowTest {
verify(ih.applyService, never()).createFromVerifiedIntent(any(), any());
}
@Test
@DisplayName("无 MQ 环境(决策①/④):签名收据是唯一事实源,十个快照成员完整落 fc_answer_projection")
void receiptWithoutMqLandsCompleteProjection() {
IntentHarness ih = intentHarness();
FcApplyIntent stored = intent(77, null);
stored.setFcIntentId(FC_INTENT_ID);
ih.existingIntent = stored;
Snapshot full = snapshot("{\"name\":\"张三\",\"event\":\"校运动会\"}");
full.setAnswerNo("AN-2026-001");
String receiptHash = AnswerProjectionStore.canonicalHash(full);
ContextReceiptView receipt = new ContextReceiptView();
receipt.setStatus("SUBMITTED");
receipt.setAnswerUuid(ANSWER_UUID);
receipt.setAnswerRevision(1L);
receipt.setContentHash(receiptHash);
receipt.setConsumerAppKey(APP_KEY);
receipt.setConsumerSchoolId(SCH);
receipt.setSubmitterId(1001);
receipt.setFormUuid(FORM_UUID);
receipt.setFormVersion(4);
receipt.setSnapshot(full);
ih.receipt = receipt;
IntentConfirmView done = ih.service.confirm(user(1001), 77);
assertThat(done.getApplyId()).isEqualTo(9001);
FcAnswerProjection p = ih.storedProjection;
assertThat(p.getSchId()).isEqualTo(SCH);
assertThat(p.getAnswerUuid()).isEqualTo(ANSWER_UUID);
assertThat(p.getFormUuid()).isEqualTo(FORM_UUID);
assertThat(p.getFormVersion()).isEqualTo(4);
assertThat(p.getRevision()).isEqualTo(1L);
assertThat(p.getContentHash()).isEqualTo(receiptHash);
assertThat(p.getIsDeleted()).isZero();
// 十个快照成员逐项落库(数据面镜像全文,不加工不丢弃)
assertThat(p.getAppCode()).isEqualTo("demo");
assertThat(p.getCode()).isEqualTo("honor_apply");
assertThat(p.getSubmitterId()).isEqualTo(1001);
assertThat(p.getSubjectId()).isEqualTo(1001);
assertThat(p.getStatus()).isEqualTo("draft");
assertThat(p.getValuesJson()).isEqualTo("{\"name\":\"张三\",\"event\":\"校运动会\"}");
assertThat(p.getAnswerNo()).isEqualTo("AN-2026-001");
assertThat(p.getAnswerCreateTime()).isEqualTo("2026-10-21T10:00:00");
assertThat(p.getAnswerUpdateTime()).isEqualTo("2026-10-21T10:05:00");
// 关联与受理
assertThat(p.getIntentId()).isEqualTo(77);
assertThat(p.getApplyId()).isEqualTo(9001);
assertThat(p.getLinkState()).isEqualTo(FcAnswerProjection.LINK_APPLIED);
assertThat(stored.getAnswerUuid()).isEqualTo(ANSWER_UUID);
assertThat(stored.getAnswerRevision()).isEqualTo(1L);
assertThat(stored.getContentHash()).isEqualTo(receiptHash);
assertThat(stored.getState()).isEqualTo(FcApplyIntent.STATE_APPLIED);
assertThat(stored.getConfirmedAt()).isNotNull();
}
@Test
@DisplayName("无 MQ 建件(决策①):办件固化收据快照,出站命令载荷自持同一份冻结数据,详情读固化副本")
void verifiedIntentCreatesApplyWithFrozenSnapshotAndOutboundCommand() {
FcAnswerProjection projection = projection(1L, null);
projection.setContentHash(hash(snapshot("{}")));
projection.setValuesJson("{\"name\":\"张三\",\"event\":\"校运动会\"}");
FcApplyIntent intent = intent(11, null);
ServiceApplyRepository applyRepository = mock(ServiceApplyRepository.class);
final ServiceApply[] savedApply = new ServiceApply[1];
when(applyRepository.saveAndFlush(any())).thenAnswer(call -> {
ServiceApply a = call.getArgument(0);
ReflectionTestUtils.setField(a, "id", 42);
savedApply[0] = a;
return a;
});
when(applyRepository.findById(42)).thenAnswer(call -> Optional.ofNullable(savedApply[0]));
ServiceItemRepository items = mock(ServiceItemRepository.class);
ServiceItem item = new ServiceItem();
item.setId(9);
item.setItemCode("svc_apply_01");
item.setTitle("测试事项");
item.setStatus(ServiceItemStatusEnum.PUBLISHED.getCode());
when(items.findById(9)).thenReturn(Optional.of(item));
FlowClient flowClient = mock(FlowClient.class);
when(flowClient.enabled()).thenReturn(true);
com.yunhai.svchall.repository.FcApplyIntentRepository intents =
mock(com.yunhai.svchall.repository.FcApplyIntentRepository.class);
when(intents.saveAndFlush(any())).thenAnswer(call -> call.getArgument(0));
// 出站命令用真实例(spy):payloadJson/createInTransaction 的载荷自持冻结数据必须是真话
FlowApplyCommandRepository commandRepository = mock(FlowApplyCommandRepository.class);
when(commandRepository.findByApplyIdOrderByIdDesc(anyInt()))
.thenReturn(Collections.emptyList());
when(commandRepository.saveAndFlush(any())).thenAnswer(call -> {
FlowApplyCommand c = call.getArgument(0);
ReflectionTestUtils.setField(c, "id", 7);
return c;
});
FlowApplyRecoveryService flowCommands = org.mockito.Mockito.spy(new FlowApplyRecoveryService(
commandRepository, mock(ServiceApplyRepository.class), flowClient, mapper, runningTx()));
ServiceApplyService applyService = new ServiceApplyService(applyRepository, items,
mock(ServiceFormService.class), mock(AttachmentService.class), flowClient,
mock(SiteGuard.class), mock(SiteDirectory.class),
new com.yunhai.svchall.service.ApplyViewGuard(flowClient), mapper, runningTx());
applyService.setFlowCommandsForTest(flowCommands);
applyService.setFcIntentsForTest(intents);
Integer applyId = applyService.createFromVerifiedIntent(intent, projection);
assertThat(applyId).isEqualTo(42);
assertThat(savedApply[0].getSchId()).isEqualTo(SCH);
assertThat(savedApply[0].getItemCode()).isEqualTo("svc_apply_01");
assertThat(savedApply[0].getFormVersion()).isEqualTo(4);
assertThat(savedApply[0].getFlowGuid()).isEqualTo("fg-1");
// 提交时固化:办件表单数据抄自投影快照一次,此后投影更新不再回写
assertThat(savedApply[0].getFormDataJson()).isEqualTo("{\"name\":\"张三\",\"event\":\"校运动会\"}");
assertThat(savedApply[0].getItemSnapshotJson()).contains("测试事项");
assertThat(savedApply[0].getFcIntentId()).isEqualTo(11);
assertThat(savedApply[0].getFcAnswerUuid()).isEqualTo(ANSWER_UUID);
assertThat(intent.getState()).isEqualTo(FcApplyIntent.STATE_APPLIED);
assertThat(intent.getApplyId()).isEqualTo(42);
// 出站命令与办件同一短事务落库,载荷自持冻结数据(重放不回读办件行)
ArgumentCaptor<String> payload = ArgumentCaptor.forClass(String.class);
verify(flowCommands).createInTransaction(any(), eq(11), payload.capture());
assertThat(payload.getValue()).contains("校运动会").contains("测试事项");
verify(flowCommands).dispatchAfterCommit(7);
// 展示(决策①"详情可看"):详情 VO 读的是办件行里的固化副本
com.yunhai.svchall.model.svc.SvcModels.ApplyVo vo = applyService.detailOfApplyId(42);
assertThat(vo).isNotNull();
assertThat(vo.getFormDataJson()).isEqualTo("{\"name\":\"张三\",\"event\":\"校运动会\"}");
assertThat(vo.getFormVersion()).isEqualTo(4);
}
// ==================== 出站命令与恢复(C.1/C.2) ====================
@Test
@@ -539,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());
@@ -548,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));
}
@@ -566,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());
@@ -576,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
@@ -630,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);
@@ -638,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) {
@@ -911,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 -> {
@@ -979,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<>();
@@ -0,0 +1,47 @@
package com.yunhai.svchall.service.impl;
import com.yunhai.svchall.config.FcAnswerMqConfiguration;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* SH 的 MQ 消费入口默认关闭(2026-10-22 决策④:保留实现,不为技术统一而部署)。
*
* <p>关闭必须落在<b>启动期</b>而不是运行期判断:{@code @RabbitListener} 注解处理器在容器工厂缺失时
* 直接抛异常打断启动,所以消费侧两个类都要挂 {@code @ConditionalOnExpression},且 yml 默认值
* 必须是空(部署不显式注入 {@code RABBITMQ_HOST} 就没有消费入口)。这三件事缺一,"MQ 可选"
* 就退化成"MQ 必部署"。以后真要接入:给环境注入 RABBITMQ_HOST 即可,实现原样保留。</p>
*/
@DisplayName("MQ 消费入口默认关 - 条件装配与空默认(决策④)")
class ShMqEntryDefaultOffTest {
@Test
@DisplayName("消费者与拓扑配置都挂 spring.rabbitmq.host 条件:未配 broker 时 bean 不存在")
void consumerEntryIsConditionallyAssembled() {
ConditionalOnExpression consumer =
FcAnswerConsumer.class.getAnnotation(ConditionalOnExpression.class);
assertThat(consumer).as("FcAnswerConsumer 必须条件装配").isNotNull();
assertThat(consumer.value()).contains("spring.rabbitmq.host");
ConditionalOnExpression configuration =
FcAnswerMqConfiguration.class.getAnnotation(ConditionalOnExpression.class);
assertThat(configuration).as("FcAnswerMqConfiguration 必须条件装配").isNotNull();
assertThat(configuration.value()).contains("spring.rabbitmq.host");
}
@Test
@DisplayName("application.yml 默认 RABBITMQ_HOST 为空:不注入就没有消费入口(决策④默认关)")
void ymlKeepsRabbitHostEmptyByDefault() {
InputStream in = getClass().getClassLoader().getResourceAsStream("application.yml");
assertThat(in).as("classpath 上应有 application.yml").isNotNull();
String yml = new Scanner(in, StandardCharsets.UTF_8.name()).useDelimiter("\\A").next();
assertThat(yml).contains("host: ${RABBITMQ_HOST:}");
}
}
@@ -0,0 +1,53 @@
package com.yunhai.svchall.service.impl;
import com.yunhai.svchall.entity.FcAnswerProjection;
import com.yunhai.svchall.repository.FcAnswerProjectionRepository;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.springframework.data.jpa.repository.Query;
import static org.assertj.core.api.Assertions.assertThat;
/**
* 权威修复的 SQL 护栏(2026-10-22 决策③"提交时固化"的数据库侧一半)。
*
* <p>{@code repairIfUnchanged} 是对账唯一能改已落库投影的原语。办件受理后的审批证据固化在
* {@code service_apply.form_data_json}(建件时抄一次),投影行只是<b>普通投影(数据面镜像)</b>;
* 但镜像行上的关联列({@code intent_id}/{@code apply_id}/{@code link_state})与业务扩展列
* ({@code biz_ext_json})一旦被修复顺手改写,"这份答卷是哪个办件的受理依据"就断了。
* 这条 JPQL 的 SET 子句必须永远只含快照列——用注解字面量钉住,谁往里加关联列就红。</p>
*/
@DisplayName("权威修复 SQL 护栏 - repairIfUnchanged 只准刷快照列(决策③)")
class ShProjectionRepairSqlGuardTest {
@Test
@DisplayName("SET 子句只含快照列+版本自增;WHERE 带期望三元组;关联列/业务扩展列绝不出现")
void repairIfUnchangedUpdatesSnapshotColumnsOnly() throws NoSuchMethodException {
Query query = FcAnswerProjectionRepository.class
.getMethod("repairIfUnchanged", FcAnswerProjection.class, Long.class, String.class,
Integer.class)
.getAnnotation(Query.class);
assertThat(query).as("repairIfUnchanged 必须是 @Query 条件更新").isNotNull();
String jpql = query.value();
// 快照成员 + 本地写代数自增,全部在 SET 里
assertThat(jpql)
.contains("p.revision =")
.contains("p.contentHash =")
.contains("p.formVersion =")
.contains("p.isDeleted =")
.contains("p.valuesJson =")
.contains("p.projectionVersion = p.projectionVersion + 1");
// 条件更新裁决:期望三元组必须逐字出现在 WHERE
assertThat(jpql)
.contains("p.revision = :revision")
.contains("p.contentHash = :hash")
.contains("p.isDeleted = :deleted");
// 关联列与业务扩展列不在 UPDATE 里(关联列保护 uk_projection_intent 的语义,扩展列是业务侧自留地)
assertThat(jpql)
.doesNotContain("intentId")
.doesNotContain("applyId")
.doesNotContain("linkState")
.doesNotContain("bizExtJson");
}
}
@@ -75,6 +75,8 @@ import static org.mockito.Mockito.when;
class ShReconcileTaskTest {
private static final String SELF = "100018";
/** 指令签发方=表单中心(svchall.answer-reconcile.source-app-key 的测试值) */
private static final String FC = "100021";
private static final String TASK_ID = "0f1e2d3c4b5a69788796a5b4c3d2e1f0";
private static final String FORM_UUID = "f0e1d2c3b4a5968778695a4b3c2d1e0f";
private static final String ANSWER_A = "a0000000000000000000000000000001";
@@ -103,6 +105,7 @@ class ShReconcileTaskTest {
worker = new AnswerReconcileWorker(workRepository, projectionStore, projectionRepository,
formSyncClient, runningTransactionTemplate(), mapper, SELF);
ReflectionTestUtils.setField(worker, "enabled", true);
ReflectionTestUtils.setField(worker, "sourceAppKey", FC);
ReflectionTestUtils.setField(worker, "leaseSeconds", 900);
ReflectionTestUtils.setField(worker, "maxRepairRetries", 2);
ReflectionTestUtils.setField(worker, "maxErrorItems", 5);
@@ -112,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;
});
}
// ==================== 接收与执行解耦 ====================
@@ -122,7 +131,7 @@ class ShReconcileTaskTest {
void acceptedInstructionIsPersistedAsQueuedOnly() {
when(workRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.empty());
AnswerReconcileWork work = worker.accept(instruction(1), SELF);
AnswerReconcileWork work = worker.accept(instruction(1), FC);
assertThat(work.getState()).isEqualTo(AnswerReconcileWork.STATE_QUEUED);
assertThat(work.getTaskUuid()).isEqualTo(TASK_ID);
@@ -142,14 +151,14 @@ class ShReconcileTaskTest {
@DisplayName("身份不符一律拒收:验签身份不是本应用、目标应用不是本应用、协议号不符")
void forgedOrMismatchedInstructionIsRejected() {
assertThatThrownBy(() -> worker.accept(instruction(1), "999999"))
.isInstanceOf(BizException.class).hasMessageContaining("消费身份不符");
.isInstanceOf(BizException.class).hasMessageContaining("签名方");
ReconcileInstruction foreignTarget = instruction(1);
foreignTarget.setConsumerAppKey("100020");
assertThatThrownBy(() -> worker.accept(foreignTarget, SELF))
assertThatThrownBy(() -> worker.accept(foreignTarget, FC))
.isInstanceOf(BizException.class).hasMessageContaining("不是本应用");
ReconcileInstruction badProtocol = instruction(1);
badProtocol.setProtocol("fc.form-sync.v9");
assertThatThrownBy(() -> worker.accept(badProtocol, SELF))
assertThatThrownBy(() -> worker.accept(badProtocol, FC))
.isInstanceOf(BizException.class).hasMessageContaining("协议号不符");
verify(workRepository, never()).save(any(AnswerReconcileWork.class));
}
@@ -159,27 +168,27 @@ class ShReconcileTaskTest {
void instructionCoordinatesAreValidated() {
ReconcileInstruction foreignSchool = instruction(1);
foreignSchool.setConsumerSchoolId(0L);
assertThatThrownBy(() -> worker.accept(foreignSchool, SELF))
assertThatThrownBy(() -> worker.accept(foreignSchool, FC))
.isInstanceOf(BizException.class).hasMessageContaining("真实学校");
ReconcileInstruction longId = instruction(1);
longId.setTaskId("012345678901234567890123456789012345678901234567890123456789");
assertThatThrownBy(() -> worker.accept(longId, SELF))
assertThatThrownBy(() -> worker.accept(longId, FC))
.isInstanceOf(BizException.class).hasMessageContaining("36 字符");
ReconcileInstruction badWindow = instruction(1);
badWindow.setWindowStart("2026-10-21 02:00");
assertThatThrownBy(() -> worker.accept(badWindow, SELF))
assertThatThrownBy(() -> worker.accept(badWindow, FC))
.isInstanceOf(BizException.class).hasMessageContaining("任务起点");
ReconcileInstruction negativeAttempt = instruction(1);
negativeAttempt.setAttempt(-1);
assertThatThrownBy(() -> worker.accept(negativeAttempt, SELF))
assertThatThrownBy(() -> worker.accept(negativeAttempt, FC))
.isInstanceOf(BizException.class).hasMessageContaining("attempt");
ReconcileInstruction halfCoordinate = instruction(1);
halfCoordinate.setFormVersion(null);
assertThatThrownBy(() -> worker.accept(halfCoordinate, SELF))
assertThatThrownBy(() -> worker.accept(halfCoordinate, FC))
.isInstanceOf(BizException.class).hasMessageContaining("版本确切");
}
@@ -191,14 +200,14 @@ class ShReconcileTaskTest {
stored.setScannedRows(7);
when(workRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.of(stored));
AnswerReconcileWork older = worker.accept(instruction(1), SELF);
AnswerReconcileWork older = worker.accept(instruction(1), FC);
assertThat(older.getAttempt()).isEqualTo(3);
assertThat(older.getWatermark()).isEqualTo(ANSWER_A);
assertThat(older.getScannedRows()).isEqualTo(7);
verify(workRepository, never()).save(any(AnswerReconcileWork.class));
AnswerReconcileWork redispatched = worker.accept(instruction(4), SELF);
AnswerReconcileWork redispatched = worker.accept(instruction(4), FC);
assertThat(redispatched.getTaskUuid()).isEqualTo(TASK_ID);
assertThat(redispatched.getAttempt()).isEqualTo(4);
assertThat(redispatched.getWatermark()).isEqualTo(ANSWER_A);
@@ -207,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));
@@ -215,12 +224,92 @@ class ShReconcileTaskTest {
running.setScannedRows(4);
when(workRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.of(running));
AnswerReconcileWork accepted = worker.accept(instruction(2), SELF);
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
@@ -739,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() {
@@ -890,10 +1018,14 @@ class ShReconcileTaskTest {
private static TaskAnswerReply present(String answerUuid, String hash, Long revision, boolean deleted) {
TaskAnswerReply reply = new TaskAnswerReply();
reply.setTaskId(TASK_ID);
reply.setAnswerUuid(answerUuid);
// trustedReplyMatches 加严后回执必须自带消费方坐标与纪元(缺任一即拒修)
reply.setConsumerAppKey(SELF);
reply.setConsumerSchoolId(SCHOOL);
reply.setFormUuid(FORM_UUID);
reply.setFormVersion(7);
reply.setAnswerUuid(answerUuid);
reply.setPresence(FormSyncModels.PRESENCE_PRESENT);
reply.setGrantVersion(1);
SnapshotEnvelope envelope = new SnapshotEnvelope();
envelope.setProtocolVersion("fc.answer.snapshot.v2");
envelope.setEventId(TASK_ID);
@@ -973,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;
@@ -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());
}
}