diff --git a/docs/mq-local-repair-contract.md b/docs/mq-local-repair-contract.md new file mode 100644 index 0000000..853f7b8 --- /dev/null +++ b/docs/mq-local-repair-contract.md @@ -0,0 +1,46 @@ +# MQ 独立修复契约（2026-10-23，本地候选） + +主链已经部署并由用户验收。本补丁只处理 MQ，不改变 UserBase 身份模型、HTTP 收据主链、FL、B5 或对账调度。 +代码基线为 FC 核心 SHA 1717f4c47e26918e7f582d0c00ddd23b732ab98a；旧 MQ WIP 保留在原 worktree。 + +## 身份与路由 + +- `application_info.id` 是业务应用主身份；`appKey/appSecret` 是签名凭据；`appCode` 只用于部署/寻址以及现有注册查询。 +- 兼容现有 v2 协议：`sourceAppKey/consumerAppKey` 仍携带凭据键，不改名、不把业务 id 填进去、不迁移已有业务绑定。 +- source 默认引用最终 `yh.app-auth.app-key`；显式 source 与签名键不一致时拒绝。真正发布前还按本应用的部署 code 调已有 `getAppByCode`，核验注册行的正整数 id、appCode、appKey。允许 id != appKey。不可达/错配进入有界失败退避，不记 broker 成功。 +- 不使用答卷的资源 `snapshot.appCode` 推断发布者身份。该字段是资源所属应用技术标识，原样保留。 +- v2 routing key 固定为 `fc.snapshot.v2..`，不能把应用主键或 appCode 代入。拒绝能扩展 topic 段的通配符/额外分隔符与非正学校号。 +- SH 的 binding、信封消费身份与最终签名凭据必须同源；不硬编码为 100008 或 100018。 + +## 投递状态裁决 + +| 输入 | 条件 | 结果 | +|---|---|---| +| ACK | 当前代且没有已返回事实 | BROKER_ACCEPTED；不写 consumer_applied | +| 重复 ACK | 当前代已接受 | 时间戳/计数不再改写 | +| NACK / mandatory return | 当前代，包括先前已收到 ACK | FAILED 或预算耗尽 DEAD；清 broker_accepted_at，保留原因并退避 | +| ACK 带 CorrelationData.returned | 独立 return 回调未持久化也一样 | 先按 NO_ROUTE 失败处理，不能写成功 | +| 重复负回调/迟到旧代 ACK | 负回调已推进 retry_count 或授权重放操作已变化 | 幂等忽略，不清错误、不重复扣重试预算 | +| DEAD/GRANT_REVOKED | 任意 broker 回调 | 不复活 | +| 序列化/快照损坏/注册不可达 | 未实际发送 | 失败退避；同 eventId 与不可变快照恢复，不伪造消费成功 | + +回调以现有 retry_count 与重放操作键摘要标识代际，读取同一投递行时使用写锁；没有新增库列。 +这是至少一次投递，不能声称 broker/数据库的真实并发恰好一次已验证。 +发送通道必须配置 correlated confirm、publisher-returns 和 mandatory；缺少返回事实观察能力时不启用扫描发送。 +`push_allowed=0` 保留 read/subscribe 数据权，既不建新 v2 outbox 行，也不继续发失去 push 权的旧行。 + +## N-1 元数据事件 + +`fc.events.v1` 现在显式序列化为 UTF-8 JSON Message，再调用 RabbitTemplate.send；不更换共享模板的转换器，不影响 v2。 +信封仍为十字段闭集；不含 values/附件字节/凭据。messageId 与 event_id 一致，提交后使用冻结的坐标发送。 +`fc.events.enabled` 默认 false，且仍须 broker host；仅平台注入 host 不会打开已修好的事件通道。 +该元数据通知与 v2 push 授权是两个协议；没有引入新的订阅授权模型。 +保持原 best-effort 语义：转换/声明/发送失败不反向破坏答卷提交，不增加 v1 outbox 或自动重放；后续事件可以恢复。v2 才有有界可靠重试台账。 + +## B-03 及验证边界 + +B-03 原定义是共享 `/` vhost 与同一 admin 账号，不是转换器问题。独立 vhost、分权账号与 topic 前缀约束仍需环境交付。 +FC 会主动 declare 自己的 exchange；SH RabbitAdmin 会声明队列/exchange/DLX/binding。因此权限方案须包含这些实际操作，不能只写“发布方 write”就认为配置完毕。 +普通资源权限与 topic binding key 权限须分开核验；依据 [RabbitMQ 官方访问控制表](https://www.rabbitmq.com/docs/access-control)，不是本轮 broker 实测结论。 +本轮仅本地规则、真实 JSON 转换、Spring 条件装配测试。真实路由、NO_ROUTE、ACL、HMAC身份、DB并发与生产验收均 NOT RUN。 +不得开启生产 MQ、修改 push_allowed 或 B-08。请使用独立测试 vhost 完成后续环境验证。 diff --git a/src/main/java/com/yunhai/formcenter/repository/AnswerDeliveryRepository.java b/src/main/java/com/yunhai/formcenter/repository/AnswerDeliveryRepository.java index 3eac26d..f72dab9 100644 --- a/src/main/java/com/yunhai/formcenter/repository/AnswerDeliveryRepository.java +++ b/src/main/java/com/yunhai/formcenter/repository/AnswerDeliveryRepository.java @@ -4,6 +4,8 @@ import com.yunhai.formcenter.entity.AnswerDelivery; import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Lock; +import javax.persistence.LockModeType; import org.springframework.data.jpa.repository.Query; import org.springframework.data.repository.query.Param; @@ -20,13 +22,20 @@ import java.util.Optional; */ public interface AnswerDeliveryRepository extends JpaRepository { + /** 回调/发送失败按同一行串行裁决，避免 return 与 ACK 各读旧状态后互相覆盖。 */ + @Override + @Lock(LockModeType.PESSIMISTIC_WRITE) + Optional findById(Long id); + /** 回执端点定位（eventId 全局唯一，另三键做逐字复核） */ + @Lock(LockModeType.PESSIMISTIC_WRITE) Optional findByEventId(String eventId); /** 同 revision 同消费方是否已有行（入队幂等；正常由 uk_delivery_revision 兜底） */ Optional findByAnswerUuidAndAnswerRevisionAndConsumerAppKeyAndConsumerSchoolId( String answerUuid, Long answerRevision, String consumerAppKey, Long consumerSchoolId); + @Lock(LockModeType.PESSIMISTIC_WRITE) List findByAnswerUuidAndConsumerAppKeyOrderByIdAsc(String answerUuid, String consumerAppKey); /** 到点待投（PENDING 首投与 FAILED 退避重试同一条队列） */ diff --git a/src/main/java/com/yunhai/formcenter/service/AnswerDeliveryService.java b/src/main/java/com/yunhai/formcenter/service/AnswerDeliveryService.java index c7f6031..ebbf37f 100644 --- a/src/main/java/com/yunhai/formcenter/service/AnswerDeliveryService.java +++ b/src/main/java/com/yunhai/formcenter/service/AnswerDeliveryService.java @@ -1,6 +1,7 @@ package com.yunhai.formcenter.service; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.JsonNode; import com.yunhai.formcenter.entity.Answer; import com.yunhai.formcenter.entity.AnswerDelivery; import com.yunhai.formcenter.model.AnswerDeliveryModels; @@ -22,6 +23,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.core.ReturnedMessage; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.ObjectProvider; @@ -38,10 +40,13 @@ import org.springframework.transaction.support.TransactionTemplate; import javax.annotation.PostConstruct; import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.ArrayList; import java.util.Arrays; +import java.util.Base64; import java.util.List; import java.util.UUID; import java.util.concurrent.ExecutorService; @@ -74,6 +79,7 @@ public class AnswerDeliveryService { private static final DateTimeFormatter OCCURRED_AT = DateTimeFormatter.ISO_OFFSET_DATE_TIME; private static final DateTimeFormatter TS = DateTimeFormatter.ISO_LOCAL_DATE_TIME; private static final String HEADER_DELIVERY_ID = "fc-delivery-id"; + private static final String HEADER_DELIVERY_ATTEMPT = "fc-delivery-attempt"; /** 退避序列：base×2^n，封顶 cap（broker 故障时不许热循环） */ private static final List SCANNABLE_STATES = Arrays.asList(AnswerDelivery.STATE_PENDING, AnswerDelivery.STATE_FAILED); @@ -87,9 +93,18 @@ public class AnswerDeliveryService { private final ObjectProvider templates; /** 发布方身份（服务端配置；不接受消费方申报，也不从资源 appCode 推） */ - @Value("${fc.answer.delivery.source-app-key:${spring.application.name:form-center}}") + @Value("${fc.answer.delivery.source-app-key:${yh.app-auth.app-key:}}") private String sourceAppKey; + @Value("${yh.app-auth.app-key:}") + private String signingAppKey; + + @Autowired + private com.yunhai.platform.openapi.base.OpenApiAppClient appRegistryClient; + + @Value("${spring.application.name:form-center}") + private String applicationCode; + @Value("${spring.rabbitmq.host:}") private String host; @@ -97,6 +112,12 @@ public class AnswerDeliveryService { @Value("${spring.rabbitmq.publisher-confirm-type:none}") private String confirmType; + @Value("${spring.rabbitmq.publisher-returns:false}") + private boolean publisherReturns = true; + + @Value("${spring.rabbitmq.template.mandatory:false}") + private boolean mandatory = true; + @Value("${fc.answer.delivery.batch-size:50}") private int batchSize; @@ -138,11 +159,15 @@ public class AnswerDeliveryService { @PostConstruct public void logState() { + if (isBlank(signingAppKey) || isBlank(sourceAppKey) || !signingAppKey.trim().equals(sourceAppKey.trim())) { + throw new IllegalStateException("fc.answer.delivery.source-app-key 必须等于 yh.app-auth.app-key 签名身份"); + } + sourceAppKey = sourceAppKey.trim(); if (!hostConfigured()) { log.info("[v2投递] 未配置 RABBITMQ_HOST：outbox 仍照常落库并留在 PENDING（不丢事件，B.4）"); return; } - if (!"correlated".equalsIgnoreCase(confirmType)) { + if (!"correlated".equalsIgnoreCase(confirmType) || !publisherReturns || !mandatory) { log.error("[v2投递] 已配置 broker 但 spring.rabbitmq.publisher-confirm-type={} —— " + "拿不到逐条 confirm 就不发送（宁可积压也不按未确认投递），请置 correlated", confirmType); return; @@ -159,6 +184,7 @@ public class AnswerDeliveryService { /** 投递通道是否可用（broker + confirm 双条件；不满足=只入队不发送） */ public boolean sendEnabled() { return hostConfigured() && "correlated".equalsIgnoreCase(confirmType) + && publisherReturns && mandatory && templates.getIfAvailable() != null; } @@ -340,21 +366,24 @@ public class AnswerDeliveryService { markFailed(row, "容器内无 RabbitTemplate"); return false; } + final String publishKey = callbackKey(row); try { + requireSourceApplication(row); declareExchangeOnce(template); byte[] body = objectMapper.writeValueAsString(envelopeOf(row)).getBytes(StandardCharsets.UTF_8); MessageProperties props = new MessageProperties(); props.setMessageId(row.getEventId()); - props.setCorrelationId(String.valueOf(row.getId())); + props.setCorrelationId(publishKey); props.setHeader(HEADER_DELIVERY_ID, String.valueOf(row.getId())); + props.setHeader(HEADER_DELIVERY_ATTEMPT, publishKey); props.setDeliveryMode(MessageDeliveryMode.PERSISTENT); props.setContentType(MessageProperties.CONTENT_TYPE_JSON); props.setContentEncoding(StandardCharsets.UTF_8.name()); template.send(AnswerDeliveryModels.EXCHANGE, routingKey(row), new Message(body, props), - new CorrelationData(String.valueOf(row.getId()))); + new CorrelationData(publishKey)); return true; } catch (Exception e) { - markFailed(row, e.toString()); + markFailed(row, publishKey, e.toString()); log.warn("[v2投递] 发送异常，退避重试：eventId={} cause={}", row.getEventId(), e.toString()); return false; } @@ -366,9 +395,28 @@ public class AnswerDeliveryService { } public static String routingKey(String consumerAppKey, Long consumerSchoolId) { + if (consumerAppKey == null || !consumerAppKey.matches("[A-Za-z0-9_-]+") + || consumerSchoolId == null || consumerSchoolId <= 0) { + throw new IllegalArgumentException("INVALID_MQ_ROUTE_COORDINATE"); + } return "fc.snapshot.v2." + consumerAppKey + "." + consumerSchoolId; } + /** Wire *AppKey fields contain credentials' keys, never application_info.id or resource appCode. */ + private void requireSourceApplication(AnswerDelivery row) { + if (isBlank(sourceAppKey) || !sourceAppKey.equals(signingAppKey == null ? null : signingAppKey.trim()) + || !sourceAppKey.equals(row.getSourceAppKey())) { + throw new IllegalStateException("SOURCE_CREDENTIAL_MISMATCH"); + } + JsonNode application = appRegistryClient == null ? null : appRegistryClient.getAppByCode(applicationCode); + if (application == null || !application.path("id").isIntegralNumber() + || !application.path("id").canConvertToLong() || application.path("id").asLong() <= 0 + || !sourceAppKey.equals(application.path("appKey").asText()) + || !applicationCode.equals(application.path("appCode").asText())) { + throw new IllegalStateException("SOURCE_APPLICATION_MISMATCH"); + } + } + private SnapshotEnvelope envelopeOf(AnswerDelivery row) { SnapshotEnvelope envelope = new SnapshotEnvelope(); envelope.setProtocolVersion(row.getProtocolVersion()); @@ -441,7 +489,16 @@ public class AnswerDeliveryService { return; // v1 广播通道不带我们的相关键 } try { - onBrokerConfirm(Long.valueOf(correlationData.getId()), ack, cause); + String key = correlationData.getId(); + // CorrelationData retains the return even if its separate callback could not persist it. + if (correlationData.getReturned() != null) { + ReturnedMessage returned = correlationData.getReturned(); + onUnroutable(callbackDeliveryId(key), key, returned.getExchange() + "/" + + returned.getRoutingKey() + " reply=" + returned.getReplyCode() + + " " + returned.getReplyText()); + return; + } + onBrokerConfirm(callbackDeliveryId(key), key, ack, cause); } catch (Exception e) { log.warn("[v2投递] confirm 回写失败 id={}：{}", correlationData.getId(), e.toString()); } @@ -451,7 +508,10 @@ public class AnswerDeliveryService { if (id == null) { return; } - onUnroutable(id, returned.getExchange() + "/" + returned.getRoutingKey() + Object generation = returned.getMessage().getMessageProperties() + .getHeaders().get(HEADER_DELIVERY_ATTEMPT); + String key = generation == null ? legacyCallbackKey(id) : String.valueOf(generation); + onUnroutable(id, key, returned.getExchange() + "/" + returned.getRoutingKey() + " reply=" + returned.getReplyCode() + " " + returned.getReplyText()); }); } @@ -471,44 +531,94 @@ public class AnswerDeliveryService { } } - /** confirm 只落 brokerAccepted；nack 走退避（B.4"confirm 只记 brokerAccepted"） */ + /** 代际来自现有 retry_count + 授权重放操作键，无新列；失败/新重放后旧回调不能改变新轮。 */ + static String callbackKey(AnswerDelivery row) { + String replay = row.getLastReplayOperationId(); + String digest = ""; + if (replay != null && !replay.isEmpty()) { + try { + digest = Base64.getUrlEncoder().withoutPadding().encodeToString( + MessageDigest.getInstance("SHA-256").digest(replay.getBytes(StandardCharsets.UTF_8))); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 unavailable", e); + } + } + return row.getId() + ":" + (row.getRetryCount() == null ? 0 : row.getRetryCount()) + ":" + digest; + } + + private static String legacyCallbackKey(Long id) { + return id + ":0:"; // 旧无代际回调只可能确认原始首投，不能猜测后续重试/重放。 + } + + private static Long callbackDeliveryId(String key) { + int separator = key.indexOf(':'); + return Long.valueOf(separator < 0 ? key : key.substring(0, separator)); + } + + /** 兼容首投的旧包内调用；生产新投递始终携带完整代际。 */ void onBrokerConfirm(Long deliveryId, boolean ack, String cause) { + onBrokerConfirm(deliveryId, legacyCallbackKey(deliveryId), ack, cause); + } + + /** confirm 不代表可路由；同代 return 的失败事实优先，旧代 ACK/nack 不能抹掉失败或改新轮。 */ + void onBrokerConfirm(Long deliveryId, String key, boolean ack, String cause) { transactionTemplate.executeWithoutResult(tx -> { AnswerDelivery row = deliveryRepository.findById(deliveryId).orElse(null); - if (row == null) { + if (row == null || !callbackKey(row).equals(normalizeCallbackKey(key, deliveryId)) + || terminalDelivery(row) + || (ack && AnswerDelivery.STATE_BROKER_ACCEPTED.equals(row.getState()))) { return; } - if (AnswerDelivery.STATE_BROKER_ACCEPTED.equals(row.getState())) { - return; // 重复 confirm（broker 重投同一相关键）幂等吸收 - } if (ack) { row.setState(AnswerDelivery.STATE_BROKER_ACCEPTED); row.setBrokerAcceptedAt(LocalDateTime.now()); row.setLastError(null); + row.setNextRetryAt(null); } else { + row.setBrokerAcceptedAt(null); bumpRetry(row, "broker nack: " + cause); } deliveryRepository.save(row); }); } - /** mandatory 不可路由=目标私有队列没绑上（部署契约缺隔离配置），按失败退避而不是静默丢弃 */ void onUnroutable(Long deliveryId, String detail) { + onUnroutable(deliveryId, legacyCallbackKey(deliveryId), detail); + } + + /** 同代 return 即使迟于 ACK 也要保留 NO_ROUTE；重复 return 代际已失效，不能重复增加退避。 */ + void onUnroutable(Long deliveryId, String key, String detail) { transactionTemplate.executeWithoutResult(tx -> { AnswerDelivery row = deliveryRepository.findById(deliveryId).orElse(null); - if (row == null || AnswerDelivery.STATE_BROKER_ACCEPTED.equals(row.getState())) { + if (row == null || !callbackKey(row).equals(normalizeCallbackKey(key, deliveryId)) + || terminalDelivery(row)) { return; } - markFailed(row, "不可路由（mandatory return）：" + detail); + row.setBrokerAcceptedAt(null); + bumpRetry(row, "NO_ROUTE: 不可路由（mandatory return）：" + detail); deliveryRepository.save(row); }); log.warn("[v2投递] 消息不可路由：id={} {}", deliveryId, detail); } + private static String normalizeCallbackKey(String key, Long id) { + return String.valueOf(id).equals(key) ? legacyCallbackKey(id) : key; + } + + private static boolean terminalDelivery(AnswerDelivery row) { + return AnswerDelivery.STATE_DEAD.equals(row.getState()) + || AnswerDelivery.STATE_GRANT_REVOKED.equals(row.getState()); + } + private void markFailed(AnswerDelivery row, String error) { + markFailed(row, callbackKey(row), error); + } + + private void markFailed(AnswerDelivery row, String key, String error) { transactionTemplate.executeWithoutResult(tx -> { AnswerDelivery fresh = deliveryRepository.findById(row.getId()).orElse(null); - if (fresh == null) { + if (fresh == null || !callbackKey(fresh).equals(key) || terminalDelivery(fresh) + || AnswerDelivery.STATE_BROKER_ACCEPTED.equals(fresh.getState())) { return; } bumpRetry(fresh, error); diff --git a/src/main/java/com/yunhai/formcenter/service/FcEventPublisher.java b/src/main/java/com/yunhai/formcenter/service/FcEventPublisher.java index ff311db..5622595 100644 --- a/src/main/java/com/yunhai/formcenter/service/FcEventPublisher.java +++ b/src/main/java/com/yunhai/formcenter/service/FcEventPublisher.java @@ -1,12 +1,16 @@ package com.yunhai.formcenter.service; import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.ObjectMapper; import com.yunhai.formcenter.entity.Answer; import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; @@ -58,7 +62,13 @@ public class FcEventPublisher { */ private final ObjectProvider templates; - /** 是否启用的唯一判据=宿主是否注入了 RABBITMQ_HOST（yml 把它映射到 spring.rabbitmq.host） */ + // Local serialization leaves the shared RabbitTemplate converter and v2 channel unchanged. + private final ObjectMapper eventMapper = new ObjectMapper(); + + @Value("${fc.events.enabled:false}") + private boolean eventsEnabled; + + /** Explicit events switch plus broker host; independent of the v2 snapshot outbox. */ @Value("${spring.rabbitmq.host:}") private String host; @@ -69,12 +79,12 @@ public class FcEventPublisher { if (enabled()) { log.info("[MQ] 事件发布启用（exchange={}，host={}）", EXCHANGE, host); } else { - log.info("[MQ] 未配置 RABBITMQ_HOST，事件发布禁用——拉取端点即事实源，服务照常（契约 v2 §6.2 裁决 3）"); + log.info("[MQ] events.enabled=false 或未配置 broker，事件发布禁用；拉取仍是事实源"); } } public boolean enabled() { - return host != null && !host.trim().isEmpty(); + return eventsEnabled && host != null && !host.trim().isEmpty(); } // ==================== 三个发布点 ==================== @@ -123,13 +133,19 @@ public class FcEventPublisher { private void send(Envelope envelope, String action) { try { + byte[] body = eventMapper.writeValueAsBytes(envelope); RabbitTemplate template = templates.getIfAvailable(); if (template == null) { log.warn("[MQ] 已配置 RABBITMQ_HOST 但容器里没有 RabbitTemplate，跳过发布（不影响主流程）"); return; } declareExchangeOnce(template); - template.convertAndSend(EXCHANGE, routingKey(envelope.getAppCode(), action), envelope); + MessageProperties properties = new MessageProperties(); + properties.setContentType(MessageProperties.CONTENT_TYPE_JSON); + properties.setContentEncoding("UTF-8"); + properties.setMessageId(envelope.getEventId()); + properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT); + template.send(EXCHANGE, routingKey(envelope.getAppCode(), action), new Message(body, properties)); } catch (Exception e) { // 裁决 2：broker 故障绝不阻断提交。消费方以游标拉取为事实源，这条事件丢了由下一轮增量补回 log.warn("[MQ] 事件发布失败（忽略，拉取兜底）：type={} answer={} cause={}", diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index b50a53e..9e49fc0 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -148,9 +148,12 @@ form-center: # source-app-key 默认取本应用签名身份（yh.app-auth.app-key），信封里的 sourceAppKey 由服务端给， # 不接受消费方申报。未配 broker 时行留在 PENDING、不丢事件（B.4）。 fc: + events: + # Metadata notification is opt-in too; fixing conversion must not enable a production channel. + enabled: ${FC_EVENTS_ENABLED:false} answer: delivery: - source-app-key: ${YH_APP_AUTH_APP_KEY:100021} + source-app-key: ${yh.app-auth.app-key} scan-interval-ms: ${FC_DELIVERY_SCAN_INTERVAL_MS:10000} scan-initial-delay-ms: ${FC_DELIVERY_SCAN_INITIAL_DELAY_MS:30000} batch-size: ${FC_DELIVERY_BATCH_SIZE:50} diff --git a/src/test/java/com/yunhai/formcenter/service/AnswerDeliveryBrokerRaceTest.java b/src/test/java/com/yunhai/formcenter/service/AnswerDeliveryBrokerRaceTest.java new file mode 100644 index 0000000..5d0bf46 --- /dev/null +++ b/src/test/java/com/yunhai/formcenter/service/AnswerDeliveryBrokerRaceTest.java @@ -0,0 +1,229 @@ +package com.yunhai.formcenter.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.yunhai.formcenter.entity.AnswerDelivery; +import com.yunhai.formcenter.repository.AnswerDeliveryRepository; +import com.yunhai.formcenter.repository.AnswerRepository; +import com.yunhai.formcenter.model.FormAccessModels.SubscriptionGrant; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.ReturnedMessage; +import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.mockito.ArgumentCaptor; +import java.util.Collections; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.data.jpa.repository.Lock; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionTemplate; + +import javax.persistence.LockModeType; +import java.time.LocalDateTime; +import java.util.Optional; +import java.util.function.Consumer; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.verify; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; + +/** B-09 独立规则回归；不冒充真实 RabbitMQ 或真实数据库并发验收。 */ +class AnswerDeliveryBrokerRaceTest { + private AnswerDelivery row; + private AnswerDeliveryService service; + + @BeforeEach + void setUp() { + row = new AnswerDelivery(); + row.setId(7L); + row.setEventId("event"); + row.setConsumerAppKey("consumer"); + row.setState(AnswerDelivery.STATE_PENDING); + row.setRetryCount(0); + AnswerDeliveryRepository rows = mock(AnswerDeliveryRepository.class); + when(rows.findById(7L)).thenReturn(Optional.of(row)); + TransactionTemplate tx = mock(TransactionTemplate.class); + doAnswer(call -> { + Consumer action = call.getArgument(0); + action.accept(null); + return null; + }).when(tx).executeWithoutResult(any()); + @SuppressWarnings("unchecked") + ObjectProvider templates = mock(ObjectProvider.class); + service = new AnswerDeliveryService(rows, mock(AnswerRepository.class), + mock(AnswerSnapshotService.class), mock(FormConsumerAccessService.class), + new ObjectMapper(), tx, templates); + ReflectionTestUtils.setField(service, "maxRetries", 3); + ReflectionTestUtils.setField(service, "backoffBaseSeconds", 30L); + ReflectionTestUtils.setField(service, "backoffCapSeconds", 3600L); + } + + @Test + void noRouteIsNotOverwrittenByFollowingAck() { + String sent = AnswerDeliveryService.callbackKey(row); + service.onUnroutable(7L, sent, "312 NO_ROUTE"); + String error = row.getLastError(); + LocalDateTime retryAt = row.getNextRetryAt(); + service.onBrokerConfirm(7L, sent, true, null); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + assertThat(row.getLastError()).isEqualTo(error).contains("NO_ROUTE"); + assertThat(row.getRetryCount()).isEqualTo(1); + assertThat(row.getNextRetryAt()).isEqualTo(retryAt).isAfter(LocalDateTime.now()); + assertThat(row.getBrokerAcceptedAt()).isNull(); + } + + @Test + void returnAlsoWinsIfAckCallbackAcquiredTheRowFirst() { + String sent = AnswerDeliveryService.callbackKey(row); + service.onBrokerConfirm(7L, sent, true, null); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_BROKER_ACCEPTED); + service.onUnroutable(7L, sent, "312 NO_ROUTE"); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + assertThat(row.getBrokerAcceptedAt()).isNull(); + assertThat(row.getLastError()).contains("NO_ROUTE"); + assertThat(row.getRetryCount()).isEqualTo(1); + } + + @Test + void duplicateReturnAndNackDoNotDoubleChargeSameFailedAttempt() { + String sent = AnswerDeliveryService.callbackKey(row); + service.onUnroutable(7L, sent, "312 NO_ROUTE"); + service.onUnroutable(7L, sent, "duplicate"); + service.onBrokerConfirm(7L, sent, false, "late nack"); + assertThat(row.getRetryCount()).isEqualTo(1); + assertThat(row.getLastError()).contains("NO_ROUTE").doesNotContain("duplicate", "late nack"); + } + + @Test + void onlyCurrentRetryCanClearFailureAfterSuccessfulRoute() { + String first = AnswerDeliveryService.callbackKey(row); + service.onUnroutable(7L, first, "312 NO_ROUTE"); + String retry = AnswerDeliveryService.callbackKey(row); + assertThat(retry).isNotEqualTo(first); + service.onBrokerConfirm(7L, retry, true, null); + service.onUnroutable(7L, first, "late old return"); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_BROKER_ACCEPTED); + assertThat(row.getLastError()).isNull(); + assertThat(row.getNextRetryAt()).isNull(); + assertThat(row.getBrokerAcceptedAt()).isNotNull(); + assertThat(row.getConsumerAppliedAt()).isNull(); + } + + @Test + void unroutableAttemptsBackOffThenStopAtExistingBudget() { + service.onUnroutable(7L, AnswerDeliveryService.callbackKey(row), "NO_ROUTE #1"); + LocalDateTime firstRetry = row.getNextRetryAt(); + service.onUnroutable(7L, AnswerDeliveryService.callbackKey(row), "NO_ROUTE #2"); + assertThat(row.getNextRetryAt()).isAfter(firstRetry); + String last = AnswerDeliveryService.callbackKey(row); + service.onUnroutable(7L, last, "NO_ROUTE #3"); + service.onBrokerConfirm(7L, last, true, null); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_DEAD); + assertThat(row.getRetryCount()).isEqualTo(3); + assertThat(row.getNextRetryAt()).isNull(); + assertThat(row.getLastError()).contains("NO_ROUTE #3"); + } + + @Test + void oldReplayCannotConfirmOrFailNewAuthorizedOperation() { + row.setLastReplayOperationId("old:operation"); + String old = AnswerDeliveryService.callbackKey(row); + row.setLastReplayOperationId("new:operation"); + service.onBrokerConfirm(7L, old, true, null); + service.onUnroutable(7L, old, "old operation NO_ROUTE"); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_PENDING); + assertThat(row.getRetryCount()).isZero(); + } + + @Test + void multibyteReplayOperationKeepsCorrelationWithinAmqpShortStringLimit() { + StringBuilder operation = new StringBuilder(); + for (int i = 0; i < 64; i++) operation.append("重"); + row.setLastReplayOperationId(operation.toString()); + String key = AnswerDeliveryService.callbackKey(row); + assertThat(key.getBytes(java.nio.charset.StandardCharsets.UTF_8).length).isLessThan(255); + assertThat(key).startsWith("7:0:"); + } + + @Test + void revokedGrantCannotBeReopenedByBrokerCallbacks() { + String sent = AnswerDeliveryService.callbackKey(row); + row.setState(AnswerDelivery.STATE_GRANT_REVOKED); + row.setLastError("revoked"); + service.onBrokerConfirm(7L, sent, true, null); + service.onUnroutable(7L, sent, "NO_ROUTE"); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_GRANT_REVOKED); + assertThat(row.getLastError()).isEqualTo("revoked"); + } + + @Test + void productionSendAndCallbacksCarryTheCapturedRetryGeneration() { + AnswerDeliveryRepository rows = mock(AnswerDeliveryRepository.class); + when(rows.findById(7L)).thenReturn(Optional.of(row)); + row.setFormUuid("form"); + row.setConsumerSchoolId(27L); + row.setGrantVersion(3); + row.setSnapshotJson("{}"); + when(rows.findDue(any(), any())).thenReturn(Collections.singletonList(row)); + FormConsumerAccessService access = mock(FormConsumerAccessService.class); + SubscriptionGrant grant = new SubscriptionGrant(); + grant.setConsumerAppKey("consumer"); + grant.setGrantVersion(3); + grant.setPushAllowed(true); + when(access.listEnabledSubscriptions("form", 27L)).thenReturn(Collections.singletonList(grant)); + TransactionTemplate tx = mock(TransactionTemplate.class); + doAnswer(call -> { + Consumer action = call.getArgument(0); + action.accept(null); + return null; + }).when(tx).executeWithoutResult(any()); + RabbitTemplate template = mock(RabbitTemplate.class); + @SuppressWarnings("unchecked") + ObjectProvider templates = mock(ObjectProvider.class); + when(templates.getIfAvailable()).thenReturn(template); + AnswerDeliveryService sender = new AnswerDeliveryService(rows, null, null, access, + new ObjectMapper(), tx, templates); + ReflectionTestUtils.setField(sender, "sourceAppKey", "100021"); + row.setSourceAppKey("100021"); + ReflectionTestUtils.setField(sender, "signingAppKey", "100021"); + ReflectionTestUtils.setField(sender, "applicationCode", "form-center"); + com.yunhai.platform.openapi.base.OpenApiAppClient registry = mock(com.yunhai.platform.openapi.base.OpenApiAppClient.class); + when(registry.getAppByCode("form-center")).thenReturn(new ObjectMapper().createObjectNode() + .put("id", 700021).put("appKey", 100021).put("appCode", "form-center")); + ReflectionTestUtils.setField(sender, "appRegistryClient", registry); + ReflectionTestUtils.setField(sender, "host", "broker"); + ReflectionTestUtils.setField(sender, "confirmType", "correlated"); + ReflectionTestUtils.setField(sender, "maxRetries", 3); + ReflectionTestUtils.setField(sender, "backoffBaseSeconds", 30L); + ReflectionTestUtils.setField(sender, "backoffCapSeconds", 3600L); + ReflectionTestUtils.invokeMethod(sender, "registerCallbacks"); + assertThat(sender.sendDueBatch()).isEqualTo(1); + ArgumentCaptor message = ArgumentCaptor.forClass(Message.class); + ArgumentCaptor correlation = ArgumentCaptor.forClass(CorrelationData.class); + verify(template).send(anyString(), anyString(), message.capture(), correlation.capture()); + assertThat(correlation.getValue().getId()).isEqualTo("7:0:"); + assertThat(message.getValue().getMessageProperties().getHeaders().get("fc-delivery-attempt")) + .isEqualTo(correlation.getValue().getId()); + ArgumentCaptor returns = ArgumentCaptor.forClass(RabbitTemplate.ReturnsCallback.class); + ArgumentCaptor confirms = ArgumentCaptor.forClass(RabbitTemplate.ConfirmCallback.class); + verify(template).setReturnsCallback(returns.capture()); + verify(template).setConfirmCallback(confirms.capture()); + returns.getValue().returnedMessage(new ReturnedMessage(message.getValue(), 312, "NO_ROUTE", "exchange", "route")); + confirms.getValue().confirm(correlation.getValue(), true, null); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + assertThat(row.getLastError()).contains("NO_ROUTE"); + } + + @Test + void callbackReadsUseTheSameDatabaseWriteLock() throws Exception { + Lock lock = AnswerDeliveryRepository.class.getMethod("findById", Long.class).getAnnotation(Lock.class); + assertThat(lock).isNotNull(); + assertThat(lock.value()).isEqualTo(LockModeType.PESSIMISTIC_WRITE); + } +} diff --git a/src/test/java/com/yunhai/formcenter/service/AnswerDeliveryMqContractTest.java b/src/test/java/com/yunhai/formcenter/service/AnswerDeliveryMqContractTest.java new file mode 100644 index 0000000..54c9aa9 --- /dev/null +++ b/src/test/java/com/yunhai/formcenter/service/AnswerDeliveryMqContractTest.java @@ -0,0 +1,239 @@ +package com.yunhai.formcenter.service; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.yunhai.formcenter.entity.AnswerDelivery; +import com.yunhai.formcenter.model.AnswerDeliveryModels; +import com.yunhai.formcenter.model.FormAccessModels.SubscriptionGrant; +import com.yunhai.formcenter.repository.AnswerDeliveryRepository; +import com.yunhai.platform.openapi.base.OpenApiAppClient; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.ReturnedMessage; +import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionTemplate; + +import java.time.LocalDateTime; +import java.util.Collections; +import java.util.Optional; +import java.util.function.Consumer; + +import static org.assertj.core.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +/** Local state/serialization/wiring tests; no broker and no real database concurrency. */ +class AnswerDeliveryMqContractTest { + private final ObjectMapper mapper = new ObjectMapper(); + private AnswerDelivery row; + private AnswerDeliveryService service; + private RabbitTemplate template; + private OpenApiAppClient registry; + private ObjectNode application; + private RabbitTemplate.ConfirmCallback confirm; + private RabbitTemplate.ReturnsCallback returns; + + @BeforeEach + void setUp() { + row = new AnswerDelivery(); + row.setId(9L); row.setEventId("stable-event"); + row.setProtocolVersion(AnswerDeliveryModels.PROTOCOL_VERSION); row.setEventType("updated"); + row.setSourceAppKey("100025"); row.setConsumerAppKey("100018"); row.setConsumerSchoolId(27L); + row.setGrantVersion(4); row.setFormUuid("form-uuid"); row.setFormVersion(1); + row.setAnswerUuid("answer-uuid"); row.setAnswerRevision(8L); row.setContentHash("hash"); + row.setOccurredAt(LocalDateTime.of(2026, 10, 23, 10, 0)); + row.setSnapshotJson("{\"schId\":27,\"appCode\":\"business-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-1\"}}"); + AnswerDeliveryRepository rows = mock(AnswerDeliveryRepository.class); + when(rows.findById(9L)).thenReturn(Optional.of(row)); + when(rows.findDue(any(), any())).thenReturn(Collections.singletonList(row)); + FormConsumerAccessService access = mock(FormConsumerAccessService.class); + SubscriptionGrant grant = new SubscriptionGrant(); + grant.setConsumerAppKey("100018"); grant.setGrantVersion(4); grant.setPushAllowed(true); + when(access.listEnabledSubscriptions("form-uuid", 27L)).thenReturn(Collections.singletonList(grant)); + TransactionTemplate tx = mock(TransactionTemplate.class); + doAnswer(call -> { Consumer action = call.getArgument(0); action.accept(null); return null; }) + .when(tx).executeWithoutResult(any()); + template = mock(RabbitTemplate.class); + @SuppressWarnings("unchecked") ObjectProvider provider = mock(ObjectProvider.class); + when(provider.getIfAvailable()).thenReturn(template); + service = new AnswerDeliveryService(rows, null, null, access, mapper, tx, provider); + registry = mock(OpenApiAppClient.class); + application = mapper.createObjectNode().put("id", 700025).put("appKey", 100025).put("appCode", "form-center"); + when(registry.getAppByCode("form-center")).thenReturn(application); + ReflectionTestUtils.setField(service, "sourceAppKey", "100025"); + ReflectionTestUtils.setField(service, "signingAppKey", "100025"); + ReflectionTestUtils.setField(service, "applicationCode", "form-center"); + ReflectionTestUtils.setField(service, "appRegistryClient", registry); + ReflectionTestUtils.setField(service, "host", "local-test-only"); + ReflectionTestUtils.setField(service, "confirmType", "correlated"); + ReflectionTestUtils.setField(service, "batchSize", 10); + ReflectionTestUtils.setField(service, "maxRetries", 3); + ReflectionTestUtils.setField(service, "backoffBaseSeconds", 30L); + ReflectionTestUtils.setField(service, "backoffCapSeconds", 3600L); + service.logState(); + ArgumentCaptor confirms = ArgumentCaptor.forClass(RabbitTemplate.ConfirmCallback.class); + ArgumentCaptor returned = ArgumentCaptor.forClass(RabbitTemplate.ReturnsCallback.class); + verify(template).setConfirmCallback(confirms.capture()); verify(template).setReturnsCallback(returned.capture()); + confirm = confirms.getValue(); returns = returned.getValue(); + } + + @Test void routingUsesConsumerCredentialAndSchoolNotBusinessIdOrAppCode() { + assertThat(AnswerDeliveryService.routingKey("100018", 27L)).isEqualTo("fc.snapshot.v2.100018.27"); + assertThat(AnswerDeliveryService.routingKey("100008", 27L)).isEqualTo("fc.snapshot.v2.100008.27") + .isNotEqualTo("fc.snapshot.v2.100018.27"); + assertThat(AnswerDeliveryService.routingKey("700025", 27L)).isNotEqualTo(AnswerDeliveryService.routingKey(row)); + assertThatThrownBy(() -> AnswerDeliveryService.routingKey("100018.*", 27L)).isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> AnswerDeliveryService.routingKey("100018", 0L)).isInstanceOf(IllegalArgumentException.class); + } + + @Test void successfulWirePreservesAllCoordinatesAndAnswerData() throws Exception { + assertThat(service.sendDueBatch()).isEqualTo(1); + ArgumentCaptor message = ArgumentCaptor.forClass(Message.class); + verify(template).send(eq(AnswerDeliveryModels.EXCHANGE), eq("fc.snapshot.v2.100018.27"), message.capture(), any(CorrelationData.class)); + JsonNode json = mapper.readTree(message.getValue().getBody()); + assertThat(json.size()).isEqualTo(15); + assertThat(json.get("sourceAppKey").asText()).isEqualTo("100025"); + assertThat(json.get("consumerAppKey").asText()).isEqualTo("100018"); + assertThat(json.get("consumerSchoolId").asLong()).isEqualTo(27); + assertThat(json.get("eventId").asText()).isEqualTo("stable-event"); + assertThat(json.get("formUuid").asText()).isEqualTo("form-uuid"); + assertThat(json.get("formVersion").asInt()).isEqualTo(1); + assertThat(json.get("answerUuid").asText()).isEqualTo("answer-uuid"); + assertThat(json.get("answerRevision").asLong()).isEqualTo(8); + assertThat(json.get("contentHash").asText()).isEqualTo("hash"); + assertThat(json.get("snapshot")).isEqualTo(mapper.readTree(row.getSnapshotJson())); + assertThat(message.getValue().getMessageProperties().getMessageId()).isEqualTo("stable-event"); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_PENDING); // send is not a confirm + verify(registry).getAppByCode("form-center"); verify(registry, never()).getAppById(anyString()); + } + + @Test void applicationIdMayDifferFromSigningKey() { + assertThat(application.get("id").asText()).isNotEqualTo(application.get("appKey").asText()); + assertThat(service.sendDueBatch()).isEqualTo(1); + } + + @Test void sourceKeyIsNotTheApplicationPrimaryId() { + row.setSourceAppKey("700025"); + assertThat(service.sendDueBatch()).isZero(); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + assertThat(row.getLastError()).contains("SOURCE_CREDENTIAL_MISMATCH"); + verify(template, never()).send(anyString(), anyString(), any(Message.class), any(CorrelationData.class)); + } + + @Test void signingCredentialBelongingToAnotherApplicationIsRejected() { + application.put("appKey", 100099); + assertThat(service.sendDueBatch()).isZero(); + assertThat(row.getLastError()).contains("SOURCE_APPLICATION_MISMATCH"); + } + + @Test void wrongApplicationCodeOrMissingBusinessIdCannotMasqueradeAsSource() { + application.put("appCode", "svc-hall"); + assertThat(service.sendDueBatch()).isZero(); + application.put("appCode", "form-center").remove("id"); + assertThat(service.sendDueBatch()).isZero(); + assertThat(row.getLastError()).contains("SOURCE_APPLICATION_MISMATCH"); + } + + @Test void registryFailureIsRetryableAndCannotBeConfirmedByTheFailedGeneration() { + String failed = AnswerDeliveryService.callbackKey(row); + when(registry.getAppByCode("form-center")).thenThrow(new IllegalStateException("registry unavailable")); + assertThat(service.sendDueBatch()).isZero(); + confirm.confirm(new CorrelationData(failed), true, null); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + assertThat(row.getNextRetryAt()).isNotNull(); + assertThat(row.getLastError()).contains("registry unavailable"); + } + + @Test void returnedFactOnCorrelationPreventsSuccessEvenWithoutTheSeparateReturnCallback() { + assertThat(service.sendDueBatch()).isEqualTo(1); + ArgumentCaptor message = ArgumentCaptor.forClass(Message.class); + ArgumentCaptor key = ArgumentCaptor.forClass(CorrelationData.class); + verify(template).send(anyString(), anyString(), message.capture(), key.capture()); + key.getValue().setReturned(new ReturnedMessage(message.getValue(), 312, "NO_ROUTE", AnswerDeliveryModels.EXCHANGE, "wrong.route")); + confirm.confirm(key.getValue(), true, null); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + assertThat(row.getLastError()).contains("NO_ROUTE"); + assertThat(row.getBrokerAcceptedAt()).isNull(); + } + + @Test void bindingMismatchReturnThenAckKeepsFailureAndRetry() { + assertThat(service.sendDueBatch()).isEqualTo(1); + ArgumentCaptor message = ArgumentCaptor.forClass(Message.class); + ArgumentCaptor key = ArgumentCaptor.forClass(CorrelationData.class); + verify(template).send(anyString(), anyString(), message.capture(), key.capture()); + returns.returnedMessage(new ReturnedMessage(message.getValue(), 312, "NO_ROUTE", AnswerDeliveryModels.EXCHANGE, "fc.snapshot.v2.100018.27")); + confirm.confirm(key.getValue(), true, null); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + assertThat(row.getLastError()).contains("NO_ROUTE"); + assertThat(row.getRetryCount()).isEqualTo(1); + assertThat(row.getNextRetryAt()).isNotNull(); + } + + @Test void nackBeforeAckAndDuplicateNackCannotBecomeSuccess() { + CorrelationData key = new CorrelationData(AnswerDeliveryService.callbackKey(row)); + confirm.confirm(key, false, "broker rejected"); confirm.confirm(key, true, null); confirm.confirm(key, false, "duplicate"); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + assertThat(row.getRetryCount()).isEqualTo(1); + assertThat(row.getLastError()).contains("broker rejected").doesNotContain("duplicate"); + } + + @Test void sameGenerationNackAfterAckOverridesOnlyTheBrokerState() { + CorrelationData key = new CorrelationData(AnswerDeliveryService.callbackKey(row)); + confirm.confirm(key, true, null); confirm.confirm(key, false, "negative wins"); confirm.confirm(key, true, null); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + assertThat(row.getBrokerAcceptedAt()).isNull(); + assertThat(row.getConsumerAppliedAt()).isNull(); + assertThat(row.getLastError()).contains("negative wins"); + } + + @Test void duplicateAckDoesNotRewriteTheAcceptedTimestamp() { + CorrelationData key = new CorrelationData(AnswerDeliveryService.callbackKey(row)); + confirm.confirm(key, true, null); LocalDateTime first = row.getBrokerAcceptedAt(); confirm.confirm(key, true, null); + assertThat(row.getBrokerAcceptedAt()).isEqualTo(first); assertThat(row.getRetryCount()).isZero(); + } + + @Test void conversionFailureRetriesWithSameImmutableEventAndNewGeneration() throws Exception { + ObjectMapper failing = spy(new ObjectMapper()); + doThrow(new JsonProcessingException("encode failure") {}).when(failing).writeValueAsString(any()); + ReflectionTestUtils.setField(service, "objectMapper", failing); + String original = AnswerDeliveryService.callbackKey(row); + assertThat(service.sendDueBatch()).isZero(); + assertThat(row.getLastError()).contains("encode failure"); + confirm.confirm(new CorrelationData(original), true, null); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + ReflectionTestUtils.setField(service, "objectMapper", mapper); + row.setNextRetryAt(LocalDateTime.now().minusSeconds(1)); + assertThat(service.sendDueBatch()).isEqualTo(1); + ArgumentCaptor retry = ArgumentCaptor.forClass(CorrelationData.class); + verify(template).send(anyString(), anyString(), any(Message.class), retry.capture()); + assertThat(retry.getValue().getId()).isNotEqualTo(original); + confirm.confirm(retry.getValue(), true, null); + assertThat(row.getEventId()).isEqualTo("stable-event"); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_BROKER_ACCEPTED); + assertThat(row.getLastError()).isNull(); assertThat(row.getConsumerAppliedAt()).isNull(); + } + + @Test void corruptSnapshotCannotReachBroker() { + row.setSnapshotJson("{broken"); assertThat(service.sendDueBatch()).isZero(); + assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED); + verify(template, never()).send(anyString(), anyString(), any(Message.class), any(CorrelationData.class)); + } + + @Test void returnsAndMandatoryAreRequiredBeforeTheScannerCanSend() { + assertThat(service.sendEnabled()).isTrue(); + ReflectionTestUtils.setField(service, "publisherReturns", false); assertThat(service.sendEnabled()).isFalse(); + ReflectionTestUtils.setField(service, "publisherReturns", true); + ReflectionTestUtils.setField(service, "mandatory", false); assertThat(service.sendEnabled()).isFalse(); + } +} diff --git a/src/test/java/com/yunhai/formcenter/service/AnswerDeliverySourceIdentityTest.java b/src/test/java/com/yunhai/formcenter/service/AnswerDeliverySourceIdentityTest.java new file mode 100644 index 0000000..405d20f --- /dev/null +++ b/src/test/java/com/yunhai/formcenter/service/AnswerDeliverySourceIdentityTest.java @@ -0,0 +1,53 @@ +package com.yunhai.formcenter.service; + +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.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; + +/** B-05 只证明最终配置解析与错配拒绝；没有核实 160 的实际签名配置。 */ +class AnswerDeliverySourceIdentityTest { + @Test + void deliverySourceTracksFinalSigningPropertyInsteadOfSeparateEnvironmentFallback() throws Exception { + StandardEnvironment env = new StandardEnvironment(); + new YamlPropertySourceLoader().load("application", new ClassPathResource("application.yml")) + .forEach(env.getPropertySources()::addLast); + env.getPropertySources().addFirst(new MapPropertySource("platform-final-signing-property", + Collections.singletonMap("yh.app-auth.app-key", "100008"))); + assertThat(env.getProperty("fc.answer.delivery.source-app-key")).isEqualTo("100008"); + } + + @Test + void mismatchedExplicitSourceFailsEvenWithoutBroker() { + AnswerDeliveryService service = service("100021", "100008"); + assertThatThrownBy(service::logState).isInstanceOf(IllegalStateException.class) + .hasMessageContaining("source-app-key", "yh.app-auth.app-key"); + } + + @Test + void matchingFinalSigningIdentityDoesNotNeedBroker() { + assertThatCode(service("100008", "100008")::logState).doesNotThrowAnyException(); + } + + @Test + void missingSigningIdentityFailsClosed() { + assertThatThrownBy(service("100008", "")::logState).isInstanceOf(IllegalStateException.class); + } + + private AnswerDeliveryService service(String source, String signing) { + AnswerDeliveryService service = new AnswerDeliveryService(null, null, null, null, null, null, null); + ReflectionTestUtils.setField(service, "sourceAppKey", source); + ReflectionTestUtils.setField(service, "signingAppKey", signing); + ReflectionTestUtils.setField(service, "host", ""); + return service; + } +} diff --git a/src/test/java/com/yunhai/formcenter/service/FcEventConverterContractTest.java b/src/test/java/com/yunhai/formcenter/service/FcEventConverterContractTest.java new file mode 100644 index 0000000..47b86a0 --- /dev/null +++ b/src/test/java/com/yunhai/formcenter/service/FcEventConverterContractTest.java @@ -0,0 +1,37 @@ +package com.yunhai.formcenter.service; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.amqp.support.converter.SimpleMessageConverter; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** N-1 legacy-converter evidence; production now sends explicit JSON bytes. No broker acceptance claim. */ +class FcEventConverterContractTest { + @Test + void defaultRabbitTemplateCannotEncodeTheCurrentNonSerializableEnvelope() { + RabbitTemplate template = new RabbitTemplate(); + assertThat(template.getMessageConverter()).isInstanceOf(SimpleMessageConverter.class); + assertThatThrownBy(() -> template.getMessageConverter() + .toMessage(new FcEventPublisher.Envelope(), new MessageProperties())) + .isInstanceOf(IllegalArgumentException.class).hasMessageContaining("Serializable"); + } + + @Test + void jsonConverterIsASeparatePossibleFixAndKeepsClosedTenFields() throws Exception { + ObjectMapper mapper = new ObjectMapper(); + FcEventPublisher.Envelope envelope = new FcEventPublisher.Envelope(); + envelope.setType("fc.answer.created"); + Message encoded = new Jackson2JsonMessageConverter(mapper).toMessage(envelope, new MessageProperties()); + JsonNode json = mapper.readTree(encoded.getBody()); + assertThat(json.size()).isEqualTo(10); + assertThat(json.get("type").asText()).isEqualTo("fc.answer.created"); + assertThat(encoded.getMessageProperties().getContentType()).isEqualTo(MessageProperties.CONTENT_TYPE_JSON); + } +} diff --git a/src/test/java/com/yunhai/formcenter/service/FcEventJsonPublishTest.java b/src/test/java/com/yunhai/formcenter/service/FcEventJsonPublishTest.java new file mode 100644 index 0000000..0b82b68 --- /dev/null +++ b/src/test/java/com/yunhai/formcenter/service/FcEventJsonPublishTest.java @@ -0,0 +1,127 @@ +package com.yunhai.formcenter.service; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.yunhai.formcenter.entity.Answer; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import java.util.ArrayList; +import java.util.List; + +import static org.assertj.core.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +/** Real JSON encoding and template boundary, no RabbitMQ connection. v1 remains best effort. */ +class FcEventJsonPublishTest { + private FcEventPublisher publisher; + private RabbitTemplate template; + private ObjectProvider provider; + private Answer answer; + private final ObjectMapper mapper = new ObjectMapper(); + + @BeforeEach + @SuppressWarnings("unchecked") + void setUp() { + template = spy(new RabbitTemplate()); + doReturn(null).when(template).execute(any()); + doNothing().when(template).send(anyString(), anyString(), any(Message.class)); + provider = mock(ObjectProvider.class); when(provider.getIfAvailable()).thenReturn(template); + publisher = new FcEventPublisher(provider); + ReflectionTestUtils.setField(publisher, "host", "local-test-only"); + ReflectionTestUtils.setField(publisher, "eventsEnabled", true); + answer = new Answer(); answer.setUuid("answer-uuid"); answer.setFormUuid("form-uuid"); + answer.setSchId(27L); answer.setAppCode("svc-hall"); answer.setCode("leave"); + answer.setVersion(1); answer.setAnswerNo("LEAVE-1"); answer.setValuesJson("{\"secret\":\"must-not-broadcast\"}"); + } + + @Test void hostInjectionDoesNotEnableTheFixedNotificationChannel() { + FcEventPublisher disabled = new FcEventPublisher(provider); + ReflectionTestUtils.setField(disabled, "host", "platform-host"); + assertThat(disabled.enabled()).isFalse(); disabled.answerCreated(answer); + verifyNoInteractions(provider); + } + + @Test void actualSendWorksWithDefaultConverterAndPreservesTheTenFieldContract() throws Exception { + publisher.answerCreated(answer); + ArgumentCaptor wire = ArgumentCaptor.forClass(Message.class); + verify(template).send(eq(FcEventPublisher.EXCHANGE), eq("fc.answer.svc-hall.created"), wire.capture()); + assertThat(template.getMessageConverter()).isInstanceOf(SimpleMessageConverter.class); + JsonNode json = mapper.readTree(wire.getValue().getBody()); + List keys = new ArrayList<>(); json.fieldNames().forEachRemaining(keys::add); + assertThat(keys).containsExactlyInAnyOrder("event_id", "type", "occurred_at", "app_code", "sch_id", + "form_uuid", "form_code", "answer_uuid", "version", "answer_no"); + assertThat(json.get("event_id").asText()).isNotEmpty(); + assertThat(json.get("type").asText()).isEqualTo("fc.answer.created"); + assertThat(json.get("occurred_at").asText()).contains("+08:00"); + assertThat(json.get("app_code").asText()).isEqualTo("svc-hall"); + assertThat(json.get("sch_id").asLong()).isEqualTo(27); + assertThat(json.get("form_uuid").asText()).isEqualTo("form-uuid"); + assertThat(json.get("form_code").asText()).isEqualTo("leave"); + assertThat(json.get("answer_uuid").asText()).isEqualTo("answer-uuid"); + assertThat(json.get("version").asInt()).isEqualTo(1); + assertThat(json.get("answer_no").asText()).isEqualTo("LEAVE-1"); + assertThat(json.toString()).doesNotContain("must-not-broadcast", "values", "appSecret"); + assertThat(wire.getValue().getMessageProperties().getMessageId()).isEqualTo(json.get("event_id").asText()); + assertThat(wire.getValue().getMessageProperties().getContentType()).isEqualTo(MessageProperties.CONTENT_TYPE_JSON); + assertThat(wire.getValue().getMessageProperties().getContentEncoding()).isEqualTo("UTF-8"); + } + + @Test void conversionFailureCannotReachTheTemplateOrBreakTheAnswerWrite() throws Exception { + ObjectMapper broken = spy(new ObjectMapper()); + doThrow(new JsonProcessingException("encode failure") {}).when(broken).writeValueAsBytes(any()); + ReflectionTestUtils.setField(publisher, "eventMapper", broken); + assertThatCode(() -> publisher.answerCreated(answer)).doesNotThrowAnyException(); + verifyNoInteractions(provider); assertThat(answer.getUuid()).isEqualTo("answer-uuid"); + assertThat(answer.getVersion()).isEqualTo(1); + } + + @Test void aFailedEncodeDoesNotPermanentlyDisableLaterNotifications() throws Exception { + ObjectMapper recovering = spy(new ObjectMapper()); + doThrow(new JsonProcessingException("first encode failed") {}).doCallRealMethod() + .when(recovering).writeValueAsBytes(any()); + ReflectionTestUtils.setField(publisher, "eventMapper", recovering); + publisher.answerUpdated(answer); publisher.answerUpdated(answer); + verify(template, times(1)).send(eq(FcEventPublisher.EXCHANGE), eq("fc.answer.svc-hall.updated"), any(Message.class)); + } + + @Test void afterCommitUsesTheFrozenAnswerCoordinates() throws Exception { + TransactionSynchronizationManager.initSynchronization(); + try { + publisher.answerCreated(answer); + verify(template, never()).send(anyString(), anyString(), any(Message.class)); + answer.setUuid("later-answer"); answer.setVersion(99); answer.setAnswerNo("later-number"); + TransactionSynchronizationManager.getSynchronizations().forEach(s -> s.afterCommit()); + ArgumentCaptor wire = ArgumentCaptor.forClass(Message.class); + verify(template).send(anyString(), anyString(), wire.capture()); + JsonNode json = mapper.readTree(wire.getValue().getBody()); + assertThat(json.get("answer_uuid").asText()).isEqualTo("answer-uuid"); + assertThat(json.get("version").asInt()).isEqualTo(1); + assertThat(json.get("answer_no").asText()).isEqualTo("LEAVE-1"); + } finally { TransactionSynchronizationManager.clearSynchronization(); } + } + + @Test void failedDeclarationIsRetriedByTheNextNotification() { + doThrow(new IllegalStateException("declare failed")).doReturn(null).when(template).execute(any()); + assertThatCode(() -> publisher.answerCreated(answer)).doesNotThrowAnyException(); + publisher.answerCreated(answer); + verify(template, times(2)).execute(any()); verify(template).send(anyString(), anyString(), any(Message.class)); + } + + @Test void sendExceptionRemainsBestEffortAndNeverClaimsDelivery() { + doThrow(new IllegalStateException("send failed")).when(template).send(anyString(), anyString(), any(Message.class)); + assertThatCode(() -> publisher.answerTombstoned(answer)).doesNotThrowAnyException(); + verify(template, times(1)).send(anyString(), eq("fc.answer.svc-hall.tombstoned"), any(Message.class)); + assertThat(answer.getUuid()).isEqualTo("answer-uuid"); + } +} diff --git a/src/test/java/com/yunhai/formcenter/service/FcEventPublisherTest.java b/src/test/java/com/yunhai/formcenter/service/FcEventPublisherTest.java index baec9d6..4f5ed89 100644 --- a/src/test/java/com/yunhai/formcenter/service/FcEventPublisherTest.java +++ b/src/test/java/com/yunhai/formcenter/service/FcEventPublisherTest.java @@ -7,6 +7,7 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.mockito.ArgumentMatchers; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.core.Message; import org.springframework.beans.factory.ObjectProvider; import org.springframework.test.util.ReflectionTestUtils; import org.springframework.transaction.support.TransactionSynchronization; @@ -71,6 +72,7 @@ class FcEventPublisherTest { @SuppressWarnings("unchecked") ObjectProvider provider = mock(ObjectProvider.class); FcEventPublisher publisher = new FcEventPublisher(provider); + ReflectionTestUtils.setField(publisher, "eventsEnabled", true); ReflectionTestUtils.setField(publisher, "host", ""); assertThat(publisher.enabled()).isFalse(); @@ -88,21 +90,22 @@ class FcEventPublisherTest { ObjectProvider provider = mock(ObjectProvider.class); when(provider.getIfAvailable()).thenReturn(template); FcEventPublisher publisher = new FcEventPublisher(provider); + ReflectionTestUtils.setField(publisher, "eventsEnabled", true); ReflectionTestUtils.setField(publisher, "host", "localhost"); TransactionSynchronizationManager.initSynchronization(); try { publisher.answerCreated(answer()); - verify(template, never()).convertAndSend(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), - ArgumentMatchers.any(Object.class)); + verify(template, never()).send(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), + ArgumentMatchers.any(Message.class)); for (TransactionSynchronization sync : TransactionSynchronizationManager.getSynchronizations()) { sync.afterCommit(); } - verify(template).convertAndSend(ArgumentMatchers.eq(FcEventPublisher.EXCHANGE), + verify(template).send(ArgumentMatchers.eq(FcEventPublisher.EXCHANGE), ArgumentMatchers.eq("fc.answer.demo.created"), - ArgumentMatchers.any(FcEventPublisher.Envelope.class)); + ArgumentMatchers.any(Message.class)); } finally { TransactionSynchronizationManager.clearSynchronization(); } @@ -113,12 +116,13 @@ class FcEventPublisherTest { void sendFailureIsSwallowed() { RabbitTemplate template = mock(RabbitTemplate.class); doThrow(new IllegalStateException("broker down")).when(template) - .convertAndSend(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), - ArgumentMatchers.any(Object.class)); + .send(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), + ArgumentMatchers.any(Message.class)); @SuppressWarnings("unchecked") ObjectProvider provider = mock(ObjectProvider.class); when(provider.getIfAvailable()).thenReturn(template); FcEventPublisher publisher = new FcEventPublisher(provider); + ReflectionTestUtils.setField(publisher, "eventsEnabled", true); ReflectionTestUtils.setField(publisher, "host", "localhost"); // 无事务上下文时就地发布（同步器未激活的那条分支）