Author SHA1 Message Date
chenxing 96cec3c2fc test(ci): 门禁选择器纳入 MQ 定向用例(3 类)——zcode 审查轮补齐,远端 CI 覆盖本地 97/97 的 SH 半边 2026-10-23 16:44:41 +08:00
chenxing 40dd4a1185 fix(mq): B-04 显式 mq-enabled 默认关+最终凭据链+注册应用核验+动态 binding——平台 host 注入不再自动启用 MQ 消费 2026-10-23 16:37:10 +08:00
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
28 changed files with 1407 additions and 98 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,ShMqExplicitEnableTest,ShMqRoutingIdentityTest,FcAnswerConsumerMqTest
# 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 已能发送冻结轮次身份;存量缺身份积压需单独清点,不回退新数据安全语义。
+10
View File
@@ -0,0 +1,10 @@
# SH MQ 独立配置契约(2026-10-23)
- `svchall.fc.mq-enabled` 默认 false。配置类与消费者使用相同条件;平台注入 RabbitMQ host 不自动打开能力。
- `svchall.fc.consumer-app-key` 默认引用最终 `yh.app-auth.app-key`,不再直接使用陈旧 env 回退 100018。
- 只有显式启用且配置 host 才装配 MQ。启用时校验消费凭据键与签名键一致,再用已有 UserBase `getAppByCode(spring.application.name)` 核验注册行的 id/appKey/appCode 关系。
- 业务应用主身份是 application_info.id;id 可以不等于凭据 appKey。禁止拿 id 或 appCode 替代 MQ 协议的 *AppKey 字段。
- binding 继续动态生成 `fc.snapshot.v2.<自身消费凭据键>.*`,没有硬编码 100008。空键、通配符或额外 topic 段拒绝。
- 仅默认链修正不会移除 broker 已有旧 binding;旧绑定、topic 权限及独立 vhost 清理必须由 zcode 在另行授权的环境验证。
- 当前 MQ 投影入口的 DB 异常分支直接 basicNack(requeue=false),并吞掉异常;不能只凭工厂 retryTemplate 配置宣称发生了三次 listener 重试。实际死信路由和恢复能力 NOT RUN,本补丁未重构这一策略。
- 本地 JSON 交接/ACK顺序测试不等于真实数据库提交或 broker 验证。核心 HTTP 签名收据链仍独立运行;本轮不处理 B5、不改 UserBase 模型、不启用对账。
@@ -12,6 +12,8 @@ import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFacto
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.annotation.Autowired;
import javax.annotation.PostConstruct;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -21,22 +23,22 @@ import org.springframework.retry.support.RetryTemplate;
/**
* v2 快照投递的消费侧拓扑(answer-delivery-v2 §3.3/§7,TASK_2 B.3/B.5)。
*
* <p><b>整体条件装配:只有显式配了 {@code spring.rabbitmq.host} 才存在</b>。未配 broker 时
* <p><b>整体条件装配:只有显式设置 {@code svchall.fc.mq-enabled=true} 且配置 {@code spring.rabbitmq.host} 才存在</b>。未配 broker 时
* 不建监听容器(否则后台反复重连把启动与日志拖脏),{@code FcAnswerConsumer} 上的
* {@code @RabbitListener} 因没有处理器而不生效——可信建件仍走"签名收据确认"那条路,
* 只是延迟从秒级变成"用户点确认那一刻"。</p>
* {@code @RabbitListener} 因没有处理器而不生效。平台统一注入 host 不能自动打开消费能力;
* 当前可信建件主链始终是签名收据 + HTTP,不依赖 MQ。</p>
*
* <p><b>队列隔离是这条链的硬前提</b>:完整值文档只进<b>按消费 appKey + 租户域绑定的私有队列</b>。
* routing key 形如 {@code fc.snapshot.v2.<consumerAppKey>.<schoolId>},本应用只绑自己 appKey
* 那一段({@code .*} 只覆盖自己的多租户域),<b>新消费者不得能绑其他目标的 routing key</b>
* (v2 §5 撤权边界的 broker 侧一半)。部署必须落实的实值来源见交付说明"broker 隔离配置"一节:
* 独立 vhost、账号仅对本 vhost 有 configure/write/read、禁止通配订阅他人前缀;
* 这三条没有落地就不能打开 {@code RABBITMQ_HOST}——宁可不投,也不在共享 vhost 上发明文。</p>
* 这三条没有落地就不能设置 {@code svchall.fc.mq-enabled=true}——宁可不投,也不在共享 vhost 上发明文。</p>
*/
@Slf4j
@Configuration
@org.springframework.amqp.rabbit.annotation.EnableRabbit
@ConditionalOnExpression("!'${spring.rabbitmq.host:}'.trim().isEmpty()")
@ConditionalOnExpression("${svchall.fc.mq-enabled:false} && !'${spring.rabbitmq.host:}'.trim().isEmpty()")
public class FcAnswerMqConfiguration {
/** 中心侧 v2 exchange(与 FC {@code AnswerDeliveryModels.EXCHANGE} 同一字面量) */
@@ -49,9 +51,32 @@ public class FcAnswerMqConfiguration {
public static final String DLX = "svc-hall.fc.answer.snapshot.v2.dlx";
public static final String DLQ = QUEUE + ".dlq";
@Value("${svchall.fc.consumer-app-key:${yh.app-auth.app-key:100018}}")
@Value("${svchall.fc.consumer-app-key:${yh.app-auth.app-key:}}")
private String consumerAppKey;
@Value("${yh.app-auth.app-key:}")
private String signingAppKey;
@Value("${spring.application.name:svc-hall}")
private String applicationCode;
@Autowired
private com.yunhai.platform.openapi.base.OpenApiAppClient appRegistryClient;
@PostConstruct
public void validateIdentity() {
String key = trimmedAppKey();
if (signingAppKey == null || !key.equals(signingAppKey.trim())) {
throw new IllegalStateException("CONSUMER_CREDENTIAL_MISMATCH");
}
com.fasterxml.jackson.databind.JsonNode app = appRegistryClient.getAppByCode(applicationCode);
if (app == null || !app.path("id").isIntegralNumber() || !app.path("id").canConvertToLong()
|| app.path("id").asLong() <= 0 || !key.equals(app.path("appKey").asText())
|| !applicationCode.equals(app.path("appCode").asText())) {
throw new IllegalStateException("CONSUMER_APPLICATION_MISMATCH");
}
}
@Value("${svchall.fc.prefetch-count:10}")
private int prefetchCount;
@@ -123,6 +148,10 @@ public class FcAnswerMqConfiguration {
}
private String trimmedAppKey() {
return consumerAppKey == null ? "" : consumerAppKey.trim();
String key = consumerAppKey == null ? "" : consumerAppKey.trim();
if (!key.matches("[A-Za-z0-9_-]+")) {
throw new IllegalStateException("INVALID_CONSUMER_ROUTE_KEY");
}
return key;
}
}
@@ -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);
}
@@ -26,7 +26,7 @@ import java.util.List;
/**
* v2 快照消费者(answer-delivery-v2 §7,TASK_2 B.5/B.6)。
*
* <p>本类<b>整体条件装配</b>:只有显式配了 {@code spring.rabbitmq.host} 才存在(与
* <p>本类<b>整体条件装配</b>:只有显式设置 {@code svchall.fc.mq-enabled=true} 且配置 {@code spring.rabbitmq.host} 才存在(与
* {@link FcAnswerMqConfiguration} 同一条件表达式)。未接 MQ 的环境里 bean 不创建,
* {@code @RabbitListener} 端点自然不被注册——"MQ 可选"必须落到启动期:注解处理器在
* 容器工厂缺失时会直接抛 {@code NoSuchBeanDefinitionException} 打断启动,不会优雅跳过。
@@ -49,7 +49,7 @@ import java.util.List;
@Slf4j
@Component
@org.springframework.boot.autoconfigure.condition.ConditionalOnExpression(
"!'${spring.rabbitmq.host:}'.trim().isEmpty()")
"${svchall.fc.mq-enabled:false} && !'${spring.rabbitmq.host:}'.trim().isEmpty()")
public class FcAnswerConsumer {
private final AnswerProjectionStore projectionStore;
@@ -57,7 +57,7 @@ public class FcAnswerConsumer {
private final FcAnswerInboxRepository inboxRepository;
private final ObjectMapper objectMapper;
@Value("${svchall.fc.consumer-app-key:${yh.app-auth.app-key:100018}}")
@Value("${svchall.fc.consumer-app-key:${yh.app-auth.app-key:}}")
private String consumerAppKey;
@Value("${svchall.fc.report-batch-size:20}")
@@ -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) {
+3 -1
View File
@@ -149,8 +149,10 @@ svchall:
public-base-url: http://localhost:8086/upload/
# ===== 可信建件与快照消费(answer-delivery-v2 §6~§8,TASK_2)=====
fc:
# 平台注入 RabbitMQ host 不代表 SH 允许消费;HTTP 收据建件主链不依赖 MQ。
mq-enabled: ${SVCHALL_FC_MQ_ENABLED:false}
# 本应用消费身份(默认取签名身份同一把 appKey;信封目标必须等于它,否则整封拒收)
consumer-app-key: ${YH_APP_AUTH_APP_KEY:100018}
consumer-app-key: ${yh.app-auth.app-key:}
# 被嵌填报页基址 = 表单中心<b>前端</b>地址(不是后端 API 地址):iframe 的 src 由服务端拼,
# 浏览器传进来的任何路径都不采信(A.2 末句)。缺键即拒开托管填报页——宁可不给入口,
# 也不给一个"拼了个大概能打开的地址"的入口。本地=FC front dev(8089),生产 env 注入。
@@ -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());
}
@@ -0,0 +1,94 @@
package com.yunhai.svchall.service.impl;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.yunhai.svchall.entity.FcAnswerInbox;
import com.yunhai.svchall.model.FcApplyModels.SnapshotEnvelope;
import com.yunhai.svchall.repository.FcAnswerInboxRepository;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.test.util.ReflectionTestUtils;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.Optional;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
/** Local JSON handoff and manual ACK/NACK ordering; not real broker or transaction commit evidence. */
class FcAnswerConsumerMqTest {
private final ObjectMapper mapper = new ObjectMapper();
private AnswerProjectionStore store;
private FcSubmissionClient client;
private FcAnswerInboxRepository rows;
private FcAnswerConsumer consumer;
private Channel channel;
private final String json = "{\"protocolVersion\":\"fc.answer.snapshot.v2\",\"eventId\":\"evt\","
+ "\"eventType\":\"updated\",\"sourceAppKey\":\"100025\",\"consumerAppKey\":\"100018\","
+ "\"consumerSchoolId\":27,\"grantVersion\":4,\"formUuid\":\"form\",\"formVersion\":1,"
+ "\"answerUuid\":\"answer\",\"answerRevision\":8,\"isDeleted\":false,\"contentHash\":\"hash\","
+ "\"occurredAt\":\"2026-10-23T10:00:00+08:00\",\"snapshot\":{\"schId\":27,\"appCode\":\"owner\","
+ "\"code\":\"leave\",\"submitterId\":41,\"subjectId\":42,\"status\":\"processed\","
+ "\"values\":{\"reason\":\"leave\"},\"answerNo\":\"LEAVE-8\",\"createTime\":\"2026-10-23T10:00:00\","
+ "\"updateTime\":\"2026-10-23T10:01:00\",\"intent\":{\"intentId\":\"intent\"}}}";
@BeforeEach void setUp() {
store = mock(AnswerProjectionStore.class); client = mock(FcSubmissionClient.class);
rows = mock(FcAnswerInboxRepository.class); channel = mock(Channel.class);
consumer = new FcAnswerConsumer(store, client, rows, mapper);
ReflectionTestUtils.setField(consumer, "consumerAppKey", "100018");
}
@Test void validJsonPreservesEveryAnswerCoordinateBeforeAck() throws Exception {
when(store.applySnapshot(any())).thenReturn(new AnswerProjectionStore.Disposition(
AnswerProjectionStore.Outcome.REJECTED, "evt", FcAnswerInbox.STATE_REJECTED, 8L, "hash", null, null));
consumer.onMessage(message(json), channel, 5L);
ArgumentCaptor<SnapshotEnvelope> envelope = ArgumentCaptor.forClass(SnapshotEnvelope.class);
InOrder order = inOrder(store, channel); order.verify(store).applySnapshot(envelope.capture());
order.verify(channel).basicAck(5L, false);
// Compare the complete JSON wire representation; DTO LongNode vs parsed IntNode is not data loss.
com.fasterxml.jackson.databind.JsonNode decoded = mapper.readTree(mapper.writeValueAsBytes(envelope.getValue()));
assertThat(decoded).isEqualTo(mapper.readTree(json));
verifyNoInteractions(client);
}
@Test void invalidJsonIsAckedAsPoisonWithoutPretendingItWasApplied() throws Exception {
consumer.onMessage(message("{broken"), channel, 6L);
verify(channel).basicAck(6L, false); verify(channel, never()).basicNack(anyLong(), anyBoolean(), anyBoolean());
verifyNoInteractions(store, client);
}
@Test void failedProjectionIsNackedWithoutAckOrFalseAppliedReport() throws Exception {
when(store.applySnapshot(any())).thenThrow(new IllegalStateException("database unavailable"));
consumer.onMessage(message(json), channel, 7L);
verify(channel).basicNack(7L, false, false); verify(channel, never()).basicAck(anyLong(), anyBoolean());
verifyNoInteractions(client);
}
@Test void reportFailureKeepsTheCommittedOutcomeAndSchedulesOnlyReportRetry() throws Exception {
when(store.applySnapshot(any())).thenReturn(new AnswerProjectionStore.Disposition(
AnswerProjectionStore.Outcome.APPLIED, "evt", FcAnswerInbox.STATE_APPLIED, 8L, "hash", 42, null));
FcAnswerInbox row = new FcAnswerInbox(); row.setState(FcAnswerInbox.STATE_APPLIED);
when(rows.findByEventIdAndConsumerAppKey("evt", "100018")).thenReturn(Optional.of(row));
consumer.onMessage(message(json), channel, 8L);
InOrder order = inOrder(store, channel, client);
order.verify(store).applySnapshot(any()); order.verify(channel).basicAck(8L, false);
order.verify(client).reportApplied(any());
assertThat(row.getState()).isEqualTo(FcAnswerInbox.STATE_APPLIED);
assertThat(row.getReportState()).isEqualTo(FcAnswerInbox.REPORT_FAILED);
assertThat(row.getReportAttempts()).isEqualTo(1);
assertThat(row.getNextReportAt()).isAfter(LocalDateTime.now());
verify(channel, never()).basicNack(anyLong(), anyBoolean(), anyBoolean());
}
private Message message(String body) {
MessageProperties props = new MessageProperties(); props.setContentType(MessageProperties.CONTENT_TYPE_JSON);
return new Message(body.getBytes(StandardCharsets.UTF_8), props);
}
}
@@ -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,59 @@
package com.yunhai.svchall.service.impl;
import com.yunhai.svchall.config.FcAnswerMqConfiguration;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.mock.env.MockEnvironment;
import org.springframework.test.util.ReflectionTestUtils;
import static org.assertj.core.api.Assertions.assertThat;
/** B-04 默认关闭的 Spring 条件装配回归,不启动真实 broker。 */
class ShMqExplicitEnableTest {
private final ApplicationContextRunner context = new ApplicationContextRunner()
.withUserConfiguration(FcAnswerMqConfiguration.class, FcAnswerConsumer.class);
@Test
void platformHostInjectionAloneDoesNotCreateTopologyOrConsumer() {
context.withPropertyValues("spring.rabbitmq.host=platform-broker").run(ctx -> {
assertThat(ctx).hasNotFailed();
assertThat(ctx).doesNotHaveBean(FcAnswerConsumer.class);
assertThat(ctx).doesNotHaveBean(FcAnswerMqConfiguration.class);
assertThat(ctx).doesNotHaveBean("fcAnswerContainerFactory");
});
}
@Test
void explicitFalseStillWinsWithPlatformHost() {
context.withPropertyValues("spring.rabbitmq.host=platform-broker", "svchall.fc.mq-enabled=false")
.run(ctx -> assertThat(ctx).hasNotFailed().doesNotHaveBean(FcAnswerConsumer.class)
.doesNotHaveBean(FcAnswerMqConfiguration.class));
}
@Test
void enablingWithoutBrokerDoesNotCreateConsumerOrFactory() {
context.withPropertyValues("svchall.fc.mq-enabled=true", "spring.rabbitmq.host=")
.run(ctx -> assertThat(ctx).hasNotFailed().doesNotHaveBean(FcAnswerConsumer.class)
.doesNotHaveBean("fcAnswerContainerFactory"));
}
@Test
void bothProductionConditionsRequireTheSameExplicitTrueAndHost() {
String topology = FcAnswerMqConfiguration.class.getAnnotation(ConditionalOnExpression.class).value();
String consumer = FcAnswerConsumer.class.getAnnotation(ConditionalOnExpression.class).value();
assertThat(consumer).isEqualTo(topology).contains("svchall.fc.mq-enabled:false");
MockEnvironment env = new MockEnvironment().withProperty("svchall.fc.mq-enabled", "true")
.withProperty("spring.rabbitmq.host", "broker");
assertThat(new SpelExpressionParser().parseExpression(env.resolvePlaceholders(topology))
.getValue(Boolean.class)).isTrue();
}
@Test
void bindingRemainsSnapshotPrefixForOwnConsumerIdentity() {
FcAnswerMqConfiguration config = new FcAnswerMqConfiguration();
ReflectionTestUtils.setField(config, "consumerAppKey", "100018");
assertThat(config.fcSnapshotBinding().getRoutingKey()).isEqualTo("fc.snapshot.v2.100018.*");
}
}
@@ -0,0 +1,91 @@
package com.yunhai.svchall.service.impl;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.yunhai.platform.openapi.base.OpenApiAppClient;
import com.yunhai.svchall.config.FcAnswerMqConfiguration;
import org.junit.jupiter.api.Test;
import org.springframework.boot.env.YamlPropertySourceLoader;
import org.springframework.core.env.MapPropertySource;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.core.io.ClassPathResource;
import org.springframework.test.util.ReflectionTestUtils;
import java.util.Collections;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
/** Binding object and registry/config contract checks; actual broker routing is NOT RUN. */
class ShMqRoutingIdentityTest {
private final ObjectMapper mapper = new ObjectMapper();
@Test void yamlUsesTheFinalSigningPropertyInsteadOfTheStaleEnvironmentFallback() throws Exception {
StandardEnvironment env = new StandardEnvironment();
new YamlPropertySourceLoader().load("application", new ClassPathResource("application.yml"))
.forEach(env.getPropertySources()::addLast);
env.getPropertySources().addFirst(new MapPropertySource("final-platform-key",
Collections.singletonMap("yh.app-auth.app-key", "100008")));
assertThat(env.getProperty("svchall.fc.consumer-app-key")).isEqualTo("100008");
assertThat(env.getProperty("svchall.fc.mq-enabled")).isEqualTo("false");
}
@Test void matchingCredentialBuildsTheCanonical100018BindingWithoutUsingThePrimaryId() {
FcAnswerMqConfiguration config = config("100018", "100018", registered(700018, 100018, "svc-hall"));
assertThatCode(config::validateIdentity).doesNotThrowAnyException();
assertThat(config.fcSnapshotBinding().getRoutingKey()).isEqualTo("fc.snapshot.v2.100018.*");
assertThat(config.fcSnapshotBinding().getExchange()).isEqualTo("fc.answer.delivery.v2");
}
@Test void stale100018CannotSurviveA100008FinalSigningConfiguration() {
FcAnswerMqConfiguration config = config("100018", "100008", registered(700008, 100008, "svc-hall"));
assertThatThrownBy(config::validateIdentity).hasMessageContaining("CONSUMER_CREDENTIAL_MISMATCH");
}
@Test void configuredCredentialDeterminesBindingWithoutHardcoding100008() {
FcAnswerMqConfiguration config = config("100008", "100008", registered(700008, 100008, "svc-hall"));
config.validateIdentity();
assertThat(config.fcSnapshotBinding().getRoutingKey()).isEqualTo("fc.snapshot.v2.100008.*")
.isNotEqualTo("fc.snapshot.v2.100018.*");
}
@Test void aCredentialRegisteredToAnotherBusinessApplicationIsRejected() {
assertThatThrownBy(config("100018", "100018", registered(700025, 100025, "form-center"))::validateIdentity)
.hasMessageContaining("CONSUMER_APPLICATION_MISMATCH");
}
@Test void businessPrimaryIdIsNeverComparedEqualToOrSubstitutedForTheCredential() {
assertThatThrownBy(config("700018", "700018", registered(700018, 100018, "svc-hall"))::validateIdentity)
.hasMessageContaining("CONSUMER_APPLICATION_MISMATCH");
}
@Test void emptyRegistryOrMissingPrimaryIdFailsClosedOnlyWhenMqIsExplicitlyEnabled() {
assertThatThrownBy(config("100018", "100018", null)::validateIdentity)
.hasMessageContaining("CONSUMER_APPLICATION_MISMATCH");
ObjectNode incomplete = registered(700018, 100018, "svc-hall"); incomplete.remove("id");
assertThatThrownBy(config("100018", "100018", incomplete)::validateIdentity)
.hasMessageContaining("CONSUMER_APPLICATION_MISMATCH");
}
@Test void routeWildcardsOrExtraWordsCannotExpandTheConsumersBinding() {
for (String key : new String[]{"", "100018.*", "100018.#", "100018.27"}) {
FcAnswerMqConfiguration config = config(key, key, null);
assertThatThrownBy(config::fcSnapshotBinding).hasMessageContaining("INVALID_CONSUMER_ROUTE_KEY");
}
}
private ObjectNode registered(int id, int credential, String code) {
return mapper.createObjectNode().put("id", id).put("appKey", credential).put("appCode", code);
}
private FcAnswerMqConfiguration config(String consumer, String signing, ObjectNode application) {
FcAnswerMqConfiguration config = new FcAnswerMqConfiguration();
ReflectionTestUtils.setField(config, "consumerAppKey", consumer);
ReflectionTestUtils.setField(config, "signingAppKey", signing);
ReflectionTestUtils.setField(config, "applicationCode", "svc-hall");
OpenApiAppClient registry = mock(OpenApiAppClient.class);
when(registry.getAppByCode("svc-hall")).thenReturn(application);
ReflectionTestUtils.setField(config, "appRegistryClient", registry);
return config;
}
}
@@ -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());
}
}