diff --git a/common-core/AGENTS.md b/common-core/AGENTS.md index 0bd37d101..98ad33565 100644 --- a/common-core/AGENTS.md +++ b/common-core/AGENTS.md @@ -14,6 +14,7 @@ | 业务异常 | `src/main/java/com/baiyue/common/core/exception/BizException.java` | | 通用审计字段 | `src/main/java/com/baiyue/common/core/model/BaseEntity.java` | | MyBatis-Plus 与审计人 | `src/main/java/com/baiyue/common/core/mybatis/` | +| 有限重试与退避 | `src/main/java/com/baiyue/common/core/retry/BoundedRetryPolicy.java`；供多个服务的 Outbox 复用，不拥有持久化或投递逻辑 | | 响应契约测试 | `src/test/java/com/baiyue/common/core/ApiResponseTest.java` | ## 修改与验证 diff --git a/common-event-contract/src/main/java/com/baiyue/common/event/EventDependencyNotReadyException.java b/common-event-contract/src/main/java/com/baiyue/common/event/EventDependencyNotReadyException.java new file mode 100644 index 000000000..e7b00596c --- /dev/null +++ b/common-event-contract/src/main/java/com/baiyue/common/event/EventDependencyNotReadyException.java @@ -0,0 +1,7 @@ +package com.baiyue.common.event; + +/** Required business state has not arrived yet; retry the same event with backoff. */ +public class EventDependencyNotReadyException extends IllegalArgumentException { + public EventDependencyNotReadyException(String message) { super(message); } + public EventDependencyNotReadyException(String message, Throwable cause) { super(message, cause); } +} diff --git a/common-event-contract/src/main/java/com/baiyue/common/event/EventEnvelopeValidator.java b/common-event-contract/src/main/java/com/baiyue/common/event/EventEnvelopeValidator.java index 980c7b5b1..6d09aae10 100644 --- a/common-event-contract/src/main/java/com/baiyue/common/event/EventEnvelopeValidator.java +++ b/common-event-contract/src/main/java/com/baiyue/common/event/EventEnvelopeValidator.java @@ -75,7 +75,7 @@ public final class EventEnvelopeValidator { public static void requireValid(EventEnvelope envelope) { List errors = validate(envelope); if (!errors.isEmpty()) { - throw new IllegalArgumentException(String.join("；", errors)); + throw new EventProtocolException(String.join("；", errors)); } } diff --git a/common-event-contract/src/main/java/com/baiyue/common/event/EventProtocolException.java b/common-event-contract/src/main/java/com/baiyue/common/event/EventProtocolException.java new file mode 100644 index 000000000..7438b5507 --- /dev/null +++ b/common-event-contract/src/main/java/com/baiyue/common/event/EventProtocolException.java @@ -0,0 +1,7 @@ +package com.baiyue.common.event; + +/** An immutable event violates the consumption protocol; retrying unchanged data cannot fix it. */ +public class EventProtocolException extends IllegalArgumentException { + public EventProtocolException(String message) { super(message); } + public EventProtocolException(String message, Throwable cause) { super(message, cause); } +} diff --git a/common-web/AGENTS.md b/common-web/AGENTS.md index 019c951df..6e566731f 100644 --- a/common-web/AGENTS.md +++ b/common-web/AGENTS.md @@ -13,6 +13,7 @@ | 请求 ID 过滤器 | `src/main/java/com/baiyue/common/web/CorrelationIdFilter.java` | | W3C Trace Context | `src/main/java/com/baiyue/common/web/TraceParent.java`；严格接收 version 00，响应 `traceparent`、`X-Trace-Id` 与 MDC `traceId` 使用同一 trace ID | | Feign Trace 传播 | `FeignTraceRequestInterceptor.java`、`FeignTracePropagationAutoConfiguration.java`；仅 Feign 存在时装配，optional 依赖不向无 Feign 服务传递 | +| 可靠事件 HTTP 客户端配置 | `EventDeliveryFeignConfiguration.java`；由业务事件 Feign 客户端显式选择，在最终构建阶段固定连接/读取超时、禁止重定向和客户端内部重试，不作为全局自动配置 | | RestClient Trace 传播 | `RestClientTraceInterceptor.java`、`RestClientTracePropagationAutoConfiguration.java`；Spring 管理的 Builder 自动装配，原始 Builder 可显式复用拦截器 | | 自动配置入口 | `src/main/java/com/baiyue/common/web/CommonWebAutoConfiguration.java`；为只扫描自身包的服务注册异常处理器和 Correlation 过滤器 | | 全局异常映射与业务失败指标 | `src/main/java/com/baiyue/common/web/GlobalExceptionHandler.java`；指标 `baiyue.api.business.failures` 按有限 `code` 标记 | diff --git a/common-web/src/main/java/com/baiyue/common/web/EventDeliveryFailure.java b/common-web/src/main/java/com/baiyue/common/web/EventDeliveryFailure.java new file mode 100644 index 000000000..341bd0e36 --- /dev/null +++ b/common-web/src/main/java/com/baiyue/common/web/EventDeliveryFailure.java @@ -0,0 +1,55 @@ +package com.baiyue.common.web; + +import com.baiyue.common.core.api.ApiResponse; +import com.fasterxml.jackson.core.JsonProcessingException; +import feign.FeignException; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Set; + +/** Explicit failure classification and bounded, credential-free delivery diagnostics. */ +public class EventDeliveryFailure extends IllegalStateException { + private static final Set PERMANENT_CODES = Set.of("A0401", "A0403", "EVT_PROTOCOL"); + private final boolean permanent; + public EventDeliveryFailure(boolean permanent, String message) { + super(message); + this.permanent = permanent; + } + public static void requireSuccessful(ApiResponse response) { + requireSuccessful(response, Set.of()); + } + public static void requireSuccessful(ApiResponse response, Set additionalPermanentCodes) { + if (response == null) throw new EventDeliveryFailure(false, "Consumer returned an empty response"); + if (!response.successful()) { + throw new EventDeliveryFailure(response.code() != null && (PERMANENT_CODES.contains(response.code()) || additionalPermanentCodes.contains(response.code())), + "Consumer rejected event: code=" + response.code() + ", message=" + response.message() + + ", requestId=" + response.requestId() + ", traceId=" + response.traceId()); + } + if (response.data() == null) throw new EventDeliveryFailure(false, "Consumer ACK is missing"); + } + public static boolean isPermanent(Throwable failure) { + Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); + for (Throwable current = failure; current != null && seen.add(current); current = current.getCause()) { + if (current instanceof EventDeliveryFailure delivery && delivery.permanent) return true; + if (current instanceof FeignException http && (http.status() == 401 || http.status() == 403)) return true; + } + return false; + } + public static String diagnostic(Throwable failure, String internalToken) { + Throwable current = failure; + Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); + while (current.getCause() != null && seen.add(current) + && ("NoFallbackAvailableException".equals(current.getClass().getSimpleName()) + || "No fallback available.".equals(current.getMessage()))) current = current.getCause(); + String message; + if (current instanceof FeignException http) message = "HTTP " + http.status(); + else if (current instanceof JsonProcessingException) message = "Invalid event JSON"; + else message = current.getMessage() == null ? "unknown" : current.getMessage(); + String text = current.getClass().getSimpleName() + ": " + message; + if (internalToken != null && !internalToken.isBlank()) text = text.replace(internalToken, "***"); + text = text.replaceAll("(?i)Bearer\\s+[^\\s,;\\\"']+", "Bearer ***") + .replaceAll("(?i)((?:X-Internal-Service-Token|Authorization|access[_-]?token|refresh[_-]?token)\\s*[=:]\\s*)[^\\s,;]+", "$1***") + .replaceAll("[\\r\\n\\t]+", " "); + return text.length() > 500 ? text.substring(0, 500) : text; + } +} diff --git a/common-web/src/test/java/com/baiyue/common/web/EventDeliveryFailureTest.java b/common-web/src/test/java/com/baiyue/common/web/EventDeliveryFailureTest.java new file mode 100644 index 000000000..c62009452 --- /dev/null +++ b/common-web/src/test/java/com/baiyue/common/web/EventDeliveryFailureTest.java @@ -0,0 +1,55 @@ +package com.baiyue.common.web; + +import com.baiyue.common.core.api.ApiResponse; +import com.fasterxml.jackson.databind.ObjectMapper; +import feign.FeignException; +import feign.Request; +import feign.Response; +import org.junit.jupiter.api.Test; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.Map; +import static org.assertj.core.api.Assertions.*; + +class EventDeliveryFailureTest { + @Test void onlyExplicitProtocolOrAuthRejectionIsPermanent() { + for (String code : new String[]{"EVT_PROTOCOL", "A0401", "A0403"}) { + Throwable error = catchThrowable(() -> EventDeliveryFailure.requireSuccessful(response(code))); + assertThat(EventDeliveryFailure.isPermanent(error)).as(code).isTrue(); + } + for (String code : new String[]{"EVT_DEPENDENCY", "A0400", "A0409", "A0429", "S0503", "unknown", null}) { + Throwable error = catchThrowable(() -> EventDeliveryFailure.requireSuccessful(response(code))); + assertThat(error).isInstanceOf(EventDeliveryFailure.class); + assertThat(EventDeliveryFailure.isPermanent(error)).as(code).isFalse(); + } + assertThat(EventDeliveryFailure.isPermanent(new IllegalArgumentException("ACK mismatch"))).isFalse(); + } + @Test void httpAuthenticationFailureIsPermanentBut429And503Retry() { + for (int status : new int[]{401, 403, 429, 503}) { + var http = http(status); + assertThat(EventDeliveryFailure.isPermanent(new RuntimeException("wrapper", http))) + .isEqualTo(status == 401 || status == 403); + assertThat(EventDeliveryFailure.diagnostic(http, "secret")).contains("HTTP " + status) + .doesNotContain("remote-body", "secret"); + } + } + @Test void diagnosticsRemoveTokenHeadersSecretsAndLineBreaks() { + var failure = new IllegalStateException("secret-value\nAuthorization=Bearer abc.def.ghi; X-Internal-Service-Token=another-secret, access_token=third-secret"); + String value = EventDeliveryFailure.diagnostic(failure, "secret-value"); + assertThat(value).doesNotContain("secret-value", "abc.def.ghi", "another-secret", "third-secret", "\n"); + assertThat(EventDeliveryFailure.diagnostic(new RuntimeException("x".repeat(1000)), "")).hasSize(500); + } + @Test void malformedJsonDiagnosticNeverCopiesPersistedPayload() { + Throwable failure = catchThrowable(() -> new ObjectMapper().readTree("private-payload-value")); + assertThat(EventDeliveryFailure.diagnostic(failure, "")) + .contains("Invalid event JSON").doesNotContain("private-payload-value"); + } + private ApiResponse response(String code) { + return new ApiResponse<>(code, "consumer-message", null, "request-id", "trace-id", Instant.now()); + } + private FeignException http(int status) { + var request = Request.create(Request.HttpMethod.POST, "http://consumer/events?token=secret", Map.of(), new byte[0], StandardCharsets.UTF_8, null); + return FeignException.errorStatus("receive", Response.builder().request(request).status(status).reason("failure") + .headers(Map.of()).body("remote-body-secret", StandardCharsets.UTF_8).build()); + } +} diff --git a/finance/src/main/java/com/baiyue/finance/service/BusinessFeeSourceEventService.java b/finance/src/main/java/com/baiyue/finance/service/BusinessFeeSourceEventService.java index ee71a1255..dd3f867d6 100644 --- a/finance/src/main/java/com/baiyue/finance/service/BusinessFeeSourceEventService.java +++ b/finance/src/main/java/com/baiyue/finance/service/BusinessFeeSourceEventService.java @@ -40,6 +40,9 @@ public class BusinessFeeSourceEventService { || envelope.targetRegion() != RegionCode.CN) { throw error("FIN-EVT-001", "跨区费用事件类型、版本或区域无效"); } + if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) { + throw new FinanceEventException("FIN-EVT-002", "事件信封摘要与实际载荷不一致"); + } BusinessFeeSourceChangedV1 source = payload(envelope); validate(envelope, source); String sourceSystem = source.sourceSystem().trim().toUpperCase(); diff --git a/finance/src/main/java/com/baiyue/finance/service/WalletCommandEventService.java b/finance/src/main/java/com/baiyue/finance/service/WalletCommandEventService.java index a386fe13e..5677355b4 100644 --- a/finance/src/main/java/com/baiyue/finance/service/WalletCommandEventService.java +++ b/finance/src/main/java/com/baiyue/finance/service/WalletCommandEventService.java @@ -112,6 +112,9 @@ public class WalletCommandEventService { || envelope.sourceRegion() != RegionCode.OVERSEAS || envelope.targetRegion() != RegionCode.CN) { throw new FinanceEventException("FIN-EVT-001", "资金指令类型、版本或区域无效"); } + if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) { + throw new FinanceEventException("FIN-EVT-002", "事件信封摘要与实际载荷不一致"); + } try { WalletCommandRequestedV1 value = objectMapper.treeToValue(envelope.payload(), WalletCommandRequestedV1.class); String operation = value.operation() == null ? "" : value.operation().trim().toUpperCase(Locale.ROOT); diff --git a/finance/src/main/java/com/baiyue/finance/service/WalletResultOutboxRelay.java b/finance/src/main/java/com/baiyue/finance/service/WalletResultOutboxRelay.java index 2c787ddf8..ba02103bc 100644 --- a/finance/src/main/java/com/baiyue/finance/service/WalletResultOutboxRelay.java +++ b/finance/src/main/java/com/baiyue/finance/service/WalletResultOutboxRelay.java @@ -1,6 +1,8 @@ package com.baiyue.finance.service; import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.web.EventDeliveryFailure; +import com.fasterxml.jackson.core.JsonProcessingException; import com.baiyue.common.core.retry.BoundedRetryPolicy; import com.baiyue.finance.client.WalletResultLastlegClient; import com.fasterxml.jackson.databind.ObjectMapper; @@ -50,21 +52,25 @@ public class WalletResultOutboxRelay { "自动投递次数已耗尽（含租约超时）", eventId, worker); return; } + boolean requestStarted = false; try { EventEnvelope envelope = objectMapper.readValue(String.valueOf(row.get("payload_json")), EventEnvelope.class).requireValid(); + requestStarted = true; var response = client.receive(internalToken, envelope); - if (response == null || !response.successful() || response.data() == null) - throw new IllegalStateException("Lastleg 未确认资金结果事件"); + EventDeliveryFailure.requireSuccessful(response); response.data().requireMatches(envelope); jdbc.update("UPDATE fin_wallet_event_outbox SET status='PUBLISHED',published_at=NOW(3),locked_by=NULL,locked_at=NULL,last_error=NULL,updated_at=NOW(3) WHERE event_id=? AND status='SENDING' AND locked_by=?", eventId, worker); } catch (Exception exception) { + String error = EventDeliveryFailure.diagnostic(exception, internalToken); + boolean permanent = EventDeliveryFailure.isPermanent(exception) + || (!requestStarted && (exception instanceof JsonProcessingException || exception instanceof IllegalArgumentException)); int retries = failures + 1; - String status = policy.exhausted(retries) ? "DEAD" : "PENDING"; - LocalDateTime next = policy.exhausted(retries) ? LocalDateTime.now() : LocalDateTime.now().plus(policy.delayAfterFailure(retries)); + String status = permanent || policy.exhausted(retries) ? "DEAD" : "PENDING"; + LocalDateTime next = permanent || policy.exhausted(retries) ? LocalDateTime.now() : LocalDateTime.now().plus(policy.delayAfterFailure(retries)); jdbc.update("UPDATE fin_wallet_event_outbox SET status=?,retry_count=?,next_retry_at=?,locked_by=NULL,locked_at=NULL,last_error=?,updated_at=NOW(3) WHERE event_id=? AND status='SENDING' AND locked_by=?", - status, retries, next, safe(exception.getMessage()), eventId, worker); - log.warn("资金结果事件投递失败 eventId={}, reason={}", eventId, exception.getMessage()); + status, retries, next, safe(error), eventId, worker); + log.warn("资金结果事件投递失败 eventId={}, reason={}", eventId, error); } } diff --git a/finance/src/test/java/com/baiyue/finance/WalletCommandEventServiceTest.java b/finance/src/test/java/com/baiyue/finance/WalletCommandEventServiceTest.java index 05129261d..fab833f4e 100644 --- a/finance/src/test/java/com/baiyue/finance/WalletCommandEventServiceTest.java +++ b/finance/src/test/java/com/baiyue/finance/WalletCommandEventServiceTest.java @@ -40,6 +40,18 @@ class WalletCommandEventServiceTest { when(transactions.getTransaction(any())).thenAnswer(invocation -> mock(TransactionStatus.class)); } + @Test + void rejectsTamperedEnvelopeBeforeOpeningWalletTransaction() { + EventEnvelope valid = envelope("FREEZE", 1L, BigDecimal.TEN); + EventEnvelope tampered = new EventEnvelope(valid.eventId(), valid.eventType(), valid.eventVersion(), + valid.occurredAt(), valid.sourceRegion(), valid.targetRegion(), valid.tenantId(), valid.aggregateType(), + valid.aggregateId(), valid.aggregateVersion(), valid.handoverVersion(), valid.traceId(), + valid.correlationId(), valid.causationId(), valid.payload(), "sha256:" + "0".repeat(64)); + assertThatThrownBy(() -> service.consume(tampered)).isInstanceOf(FinanceEventException.class) + .hasMessageContaining("摘要与实际载荷不一致"); + verifyNoInteractions(jdbc, wallet); + } + @Test void freezesWalletAndPublishesCompletedEvent() { EventEnvelope envelope = envelope("FREEZE", 31L, new BigDecimal("88.50")); diff --git a/firstleg/src/main/java/com/baiyue/firstleg/controller/EventConsumptionExceptionHandler.java b/firstleg/src/main/java/com/baiyue/firstleg/controller/EventConsumptionExceptionHandler.java new file mode 100644 index 000000000..cda48dd7a --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/controller/EventConsumptionExceptionHandler.java @@ -0,0 +1,21 @@ +package com.baiyue.firstleg.controller; + +import com.baiyue.common.core.api.ApiResponse; +import com.baiyue.common.event.EventProtocolException; +import com.baiyue.common.event.EventDependencyNotReadyException; +import org.slf4j.MDC; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; +import org.springframework.web.bind.annotation.ExceptionHandler; +import org.springframework.web.bind.annotation.RestControllerAdvice; +import java.time.Instant; + +@RestControllerAdvice +@Order(Ordered.HIGHEST_PRECEDENCE) +public class EventConsumptionExceptionHandler { + @ExceptionHandler({EventProtocolException.class, EventDependencyNotReadyException.class}) + public ApiResponse event(IllegalArgumentException failure) { + String code = failure instanceof EventDependencyNotReadyException ? "EVT_DEPENDENCY" : "EVT_PROTOCOL"; + return new ApiResponse<>(code, failure.getMessage(), null, MDC.get("requestId"), MDC.get("traceId"), Instant.now()); + } +} diff --git a/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java b/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java index 34fb7eae5..40d0ef37f 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java @@ -37,7 +37,7 @@ public class InternalOverseasContainerEventController { case EventTypes.OVERSEAS_CONTAINER_UNLOADED -> consumer.consume(event); case EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED -> detainedConsumer.consume(event); case EventTypes.OVERSEAS_DELIVERY_COMPLETED -> deliveryCompletedConsumer.consume(event); - default -> throw new IllegalArgumentException("不支持的 Firstleg 业务事件类型"); + default -> throw new com.baiyue.common.event.EventProtocolException("不支持的 Firstleg 业务事件类型"); }; return ApiResponse.success(result); } diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelay.java b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelay.java index d7c99bc08..30a5964d3 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelay.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelay.java @@ -2,6 +2,8 @@ package com.baiyue.firstleg.service; import com.baiyue.common.core.api.ApiResponse; import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.web.EventDeliveryFailure; +import com.fasterxml.jackson.core.JsonProcessingException; import com.baiyue.common.core.retry.BoundedRetryPolicy; import com.baiyue.common.event.EventEnvelopeValidator; import com.baiyue.firstleg.client.LastlegEventClient; @@ -50,15 +52,19 @@ public class LoadingCompletedOutboxRelay { repository.dead(row.getId(), workerId, failures, "自动投递次数已耗尽（含租约超时）"); continue; } + boolean requestStarted = false; try { EventEnvelope event=objectMapper.readValue(row.getPayloadJson(),EventEnvelope.class); EventEnvelopeValidator.requireValid(event); + requestStarted = true; var response=client.receive(internalToken,event); requireAcknowledgement(event, response); repository.published(row.getId(),workerId,LocalDateTime.now()); } catch(Exception exception) { - String error=safeError(exception); - if (policy.exhausted(failures + 1)) repository.dead(row.getId(), workerId, failures + 1, error); + boolean permanent = EventDeliveryFailure.isPermanent(exception) + || (!requestStarted && (exception instanceof JsonProcessingException || exception instanceof IllegalArgumentException)); + String error = EventDeliveryFailure.diagnostic(exception, internalToken); + if (permanent || policy.exhausted(failures + 1)) repository.dead(row.getId(), workerId, failures + 1, error); else repository.retry(row.getId(), workerId, LocalDateTime.now().plus(policy.delayAfterFailure(failures + 1)), error); log.warn("头程业务事件投递失败: eventType={}, eventKey={}, reason={}", row.getEventType(), row.getEventKey(), error); @@ -68,33 +74,8 @@ public class LoadingCompletedOutboxRelay { private void requireAcknowledgement(EventEnvelope event, ApiResponse response) { - if(response==null) throw new IllegalStateException("Lastleg未返回响应"); - if(!response.successful()) { - throw new IllegalStateException("Lastleg拒绝头程业务事件: code=" + response.code() - + ", message=" + response.message() + ", requestId=" + response.requestId() - + ", traceId=" + response.traceId()); - } - if (response.data() == null) throw new IllegalStateException("Lastleg ACK 缺失"); + EventDeliveryFailure.requireSuccessful(response); response.data().requireMatches(event); } - private String safeError(Exception exception) { - Throwable diagnostic=diagnosticCause(exception); - String message=diagnostic.getMessage(); - String value=diagnostic.getClass().getSimpleName() - + ((message==null || message.isBlank())?"":": " + message); - if(internalToken!=null && !internalToken.isBlank()) value=value.replace(internalToken,"***"); - value=value.replaceAll("[\\r\\n\\t]+"," "); - return value.length()>500?value.substring(0,500):value; - } - - private Throwable diagnosticCause(Throwable failure) { - Throwable current=failure; - while(current.getCause()!=null && current.getCause()!=current - && ("NoFallbackAvailableException".equals(current.getClass().getSimpleName()) - || "No fallback available.".equals(current.getMessage()))) { - current=current.getCause(); - } - return current; - } } diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasContainerEventConsumer.java b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasContainerEventConsumer.java index 1eb91f8c9..e57b3fae5 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasContainerEventConsumer.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasContainerEventConsumer.java @@ -1,6 +1,8 @@ package com.baiyue.firstleg.service; import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.event.EventProtocolException; +import com.baiyue.common.event.EventDependencyNotReadyException; import com.baiyue.common.event.DeliveryAck; import com.baiyue.common.event.EventPayloadCanonicalizer; import com.baiyue.common.event.EventPayloadHash; @@ -33,23 +35,23 @@ public class OverseasContainerEventConsumer { OverseasContainerUnloadedV1 payload = payload(event); if (payload.sourceContainerBookingId() == null || payload.sourceContainerBookingId().isBlank() || !payload.sourceContainerBookingId().equals(event.aggregateId())) { - throw new IllegalArgumentException("事件 aggregateId 与 sourceContainerBookingId 不一致"); + throw new EventProtocolException("事件 aggregateId 与 sourceContainerBookingId 不一致"); } if (payload.containerNo() == null || payload.containerNo().isBlank()) { - throw new IllegalArgumentException("货柜号不能为空"); + throw new EventProtocolException("货柜号不能为空"); } if (payload.unloadingCompletedAt() == null || payload.unloadingPhotoUploadedAt() == null) { - throw new IllegalArgumentException("卸柜完成时间和卸柜照片上传时间不能为空"); + throw new EventProtocolException("卸柜完成时间和卸柜照片上传时间不能为空"); } if (payload.actualEmptyReportedAt() != null && payload.actualEmptyReportedAt().isBefore(payload.unloadingPhotoUploadedAt())) { - throw new IllegalArgumentException("实际报空时间不能早于卸柜照片上传时间"); + throw new EventProtocolException("实际报空时间不能早于卸柜照片上传时间"); } ContainerEventInboxEntity existing = inbox.selectById(event.eventId()); if (existing != null) { if (!existing.getPayloadHash().equalsIgnoreCase(event.payloadHash())) { - throw new IllegalArgumentException("相同 eventId 对应的事件摘要冲突"); + throw new EventProtocolException("相同 eventId 对应的事件摘要冲突"); } if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费"); return DeliveryAck.of(event, DeliveryAck.Outcome.DUPLICATE); @@ -79,20 +81,20 @@ public class OverseasContainerEventConsumer { private void validateEnvelope(EventEnvelope event) { event.requireValid(); if (!EventTypes.OVERSEAS_CONTAINER_UNLOADED.equals(event.eventType()) || event.eventVersion() != 1) { - throw new IllegalArgumentException("不支持的海外货柜事件"); + throw new EventProtocolException("不支持的海外货柜事件"); } if (event.sourceRegion() != RegionCode.OVERSEAS || event.targetRegion() != RegionCode.CN || !"CONTAINER_BOOKING".equals(event.aggregateType())) { - throw new IllegalArgumentException("海外货柜事件区域或聚合类型无效"); + throw new EventProtocolException("海外货柜事件区域或聚合类型无效"); } if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(event.payload()), event.payloadHash())) { - throw new IllegalArgumentException("payloadHash 与事件负载不一致"); + throw new EventProtocolException("payloadHash 与事件负载不一致"); } } private OverseasContainerUnloadedV1 payload(EventEnvelope event) { try { return objectMapper.treeToValue(event.payload(), OverseasContainerUnloadedV1.class); } - catch (Exception exception) { throw new IllegalArgumentException("海外货柜事件负载无效", exception); } + catch (Exception exception) { throw new EventProtocolException("海外货柜事件负载无效", exception); } } private ContainerEventInboxEntity inbox(EventEnvelope event) { @@ -110,12 +112,12 @@ public class OverseasContainerEventConsumer { try { bookingId = Long.parseLong(sourceContainerBookingId); } catch (NumberFormatException exception) { - throw new IllegalArgumentException("国内货柜订舱 ID 无效", exception); + throw new EventProtocolException("国内货柜订舱 ID 无效", exception); } ContainerBookingEntity value = containers.selectActiveBookingForUpdate(bookingId); - if (value == null) throw new IllegalArgumentException("头程货柜不存在或不可用"); + if (value == null) throw new EventDependencyNotReadyException("头程货柜不存在或不可用"); if (value.getContainerNo() == null || !value.getContainerNo().equals(containerNo)) { - throw new IllegalArgumentException("事件货柜号与国内订舱记录不一致"); + throw new EventProtocolException("事件货柜号与国内订舱记录不一致"); } return value; } diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java index 2834891dd..3693a8715 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java @@ -2,6 +2,8 @@ package com.baiyue.firstleg.service; import com.baomidou.mybatisplus.core.toolkit.IdWorker; import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.event.EventProtocolException; +import com.baiyue.common.event.EventDependencyNotReadyException; import com.baiyue.common.event.DeliveryAck; import com.baiyue.common.event.EventPayloadCanonicalizer; import com.baiyue.common.event.EventPayloadHash; @@ -45,12 +47,12 @@ public class OverseasDeliveryCompletedConsumer { long inboundId = positive(payload.sourceInboundId(), "来源入仓单 ID 无效"); long tenantId = positive(event.tenantId(), "客户 ID 无效"); if (!Objects.equals(event.aggregateId(), payload.sourceInboundId())) { - throw new IllegalArgumentException("事件 aggregateId 与来源入仓单 ID 不一致"); + throw new EventProtocolException("事件 aggregateId 与来源入仓单 ID 不一致"); } ContainerEventInboxEntity existing = inbox.selectById(event.eventId()); if (existing != null) { if (!existing.getPayloadHash().equalsIgnoreCase(event.payloadHash())) { - throw new IllegalArgumentException("相同 eventId 对应的事件摘要冲突"); + throw new EventProtocolException("相同 eventId 对应的事件摘要冲突"); } if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费"); return DeliveryAck.of(event, DeliveryAck.Outcome.DUPLICATE); @@ -65,7 +67,7 @@ public class OverseasDeliveryCompletedConsumer { } InboundOrderEntity inbound = required(inboundId, tenantId, payload); if (!"LOADED".equals(inbound.getProcessStatus()) && !"DELIVERED".equals(inbound.getProcessStatus())) { - throw new IllegalArgumentException("来源头程入仓单尚未装柜完成"); + throw new EventDependencyNotReadyException("来源头程入仓单尚未装柜完成"); } if ("LOADED".equals(inbound.getProcessStatus())) { LocalDateTime completedAt = LocalDateTime.ofInstant(payload.completedAt(), ZoneOffset.UTC); @@ -102,10 +104,10 @@ public class OverseasDeliveryCompletedConsumer { if (!EventTypes.OVERSEAS_DELIVERY_COMPLETED.equals(event.eventType()) || event.eventVersion() != 1 || event.sourceRegion() != RegionCode.OVERSEAS || event.targetRegion() != RegionCode.CN || !"INBOUND_ORDER".equals(event.aggregateType())) { - throw new IllegalArgumentException("不支持的海外派送完成事件"); + throw new EventProtocolException("不支持的海外派送完成事件"); } if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(event.payload()), event.payloadHash())) { - throw new IllegalArgumentException("payloadHash 与事件负载不一致"); + throw new EventProtocolException("payloadHash 与事件负载不一致"); } } @@ -114,22 +116,23 @@ public class OverseasDeliveryCompletedConsumer { OverseasDeliveryCompletedV1 value = objectMapper.treeToValue(event.payload(), OverseasDeliveryCompletedV1.class); if (value.completedAt() == null || !StringUtils.hasText(value.deliveryNo()) || !StringUtils.hasText(value.sourceOrderId()) || !StringUtils.hasText(value.sourceOrderNo()) - || !StringUtils.hasText(value.inboundCode())) throw new IllegalArgumentException("字段不完整"); + || !StringUtils.hasText(value.inboundCode())) throw new EventProtocolException("字段不完整"); return value; } catch (Exception exception) { - throw new IllegalArgumentException("海外派送完成事件负载无效", exception); + throw new EventProtocolException("海外派送完成事件负载无效", exception); } } private InboundOrderEntity required(long inboundId, long tenantId, OverseasDeliveryCompletedV1 payload) { InboundOrderEntity inbound = inboundOrders.selectActiveByIdForUpdate(inboundId); - if (inbound == null || !Objects.equals(inbound.getTenantId(), tenantId) + if (inbound == null) throw new EventDependencyNotReadyException("来源头程入仓单不存在"); + if (!Objects.equals(inbound.getTenantId(), tenantId) || !Objects.equals(inbound.getInboundCode(), payload.inboundCode()) || !Objects.equals(String.valueOf(inbound.getOrderId()), payload.sourceOrderId()) || !Objects.equals(inbound.getOrderNo(), payload.sourceOrderNo()) || (StringUtils.hasText(payload.fbaWarehouseCode()) && StringUtils.hasText(inbound.getDestinationCode()) && !Objects.equals(inbound.getDestinationCode(), payload.fbaWarehouseCode()))) { - throw new IllegalArgumentException("事件与来源头程入仓单归属不一致"); + throw new EventProtocolException("事件与来源头程入仓单归属不一致"); } return inbound; } @@ -146,6 +149,6 @@ public class OverseasDeliveryCompletedConsumer { private static long positive(String value, String message) { try { long parsed = Long.parseLong(value); if (parsed > 0) return parsed; } catch (RuntimeException ignored) { } - throw new IllegalArgumentException(message); + throw new EventProtocolException(message); } } diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumer.java b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumer.java index 8c9df62d2..261a74004 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumer.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumer.java @@ -2,6 +2,8 @@ package com.baiyue.firstleg.service; import com.baomidou.mybatisplus.core.toolkit.IdWorker; import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.event.EventProtocolException; +import com.baiyue.common.event.EventDependencyNotReadyException; import com.baiyue.common.event.DeliveryAck; import com.baiyue.common.event.EventPayloadCanonicalizer; import com.baiyue.common.event.EventPayloadHash; @@ -41,17 +43,17 @@ public class OverseasInboundOrderDetainedConsumer { long inboundId = parsePositive(payload.sourceInboundId(), "来源入仓单 ID 无效"); long tenantId = parsePositive(event.tenantId(), "客户 ID 无效"); if (!payload.sourceInboundId().equals(event.aggregateId())) { - throw new IllegalArgumentException("事件 aggregateId 与来源入仓单 ID 不一致"); + throw new EventProtocolException("事件 aggregateId 与来源入仓单 ID 不一致"); } if (payload.changedAt() == null || payload.inboundCode() == null || payload.inboundCode().isBlank() || payload.reason() == null || payload.reason().isBlank()) { - throw new IllegalArgumentException("暂扣事件缺少入仓码、原因或变更时间"); + throw new EventProtocolException("暂扣事件缺少入仓码、原因或变更时间"); } ContainerEventInboxEntity existing = inbox.selectById(event.eventId()); if (existing != null) { if (!existing.getPayloadHash().equalsIgnoreCase(event.payloadHash())) { - throw new IllegalArgumentException("相同 eventId 对应的事件摘要冲突"); + throw new EventProtocolException("相同 eventId 对应的事件摘要冲突"); } if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费"); return DeliveryAck.of(event, DeliveryAck.Outcome.DUPLICATE); @@ -87,23 +89,23 @@ public class OverseasInboundOrderDetainedConsumer { if (!EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED.equals(event.eventType()) || event.eventVersion() != 1 || event.sourceRegion() != RegionCode.OVERSEAS || event.targetRegion() != RegionCode.CN || !"INBOUND_ORDER".equals(event.aggregateType())) { - throw new IllegalArgumentException("不支持的海外入仓单暂扣事件"); + throw new EventProtocolException("不支持的海外入仓单暂扣事件"); } if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(event.payload()), event.payloadHash())) { - throw new IllegalArgumentException("payloadHash 与事件负载不一致"); + throw new EventProtocolException("payloadHash 与事件负载不一致"); } } private OverseasInboundOrderDetainedChangedV1 payload(EventEnvelope event) { try { return objectMapper.treeToValue(event.payload(), OverseasInboundOrderDetainedChangedV1.class); } - catch (Exception exception) { throw new IllegalArgumentException("海外入仓单暂扣事件负载无效", exception); } + catch (Exception exception) { throw new EventProtocolException("海外入仓单暂扣事件负载无效", exception); } } private InboundOrderEntity requiredInbound(long inboundId, long tenantId, String inboundCode) { InboundOrderEntity value = inboundOrders.selectActiveByIdForUpdate(inboundId); - if (value == null) throw new IllegalArgumentException("来源头程入仓单不存在"); + if (value == null) throw new EventDependencyNotReadyException("来源头程入仓单不存在"); if (!Objects.equals(value.getTenantId(), tenantId) || !Objects.equals(value.getInboundCode(), inboundCode)) { - throw new IllegalArgumentException("事件与来源头程入仓单归属不一致"); + throw new EventProtocolException("事件与来源头程入仓单归属不一致"); } return value; } @@ -130,7 +132,7 @@ public class OverseasInboundOrderDetainedConsumer { private long parsePositive(String value, String message) { try { long parsed = Long.parseLong(value); if (parsed > 0) return parsed; } catch (RuntimeException ignored) { } - throw new IllegalArgumentException(message); + throw new EventProtocolException(message); } private long parseOperatorId(String value) { diff --git a/firstleg/src/test/java/com/baiyue/firstleg/controller/EventConsumptionExceptionHandlerTest.java b/firstleg/src/test/java/com/baiyue/firstleg/controller/EventConsumptionExceptionHandlerTest.java new file mode 100644 index 000000000..40cbb1d2d --- /dev/null +++ b/firstleg/src/test/java/com/baiyue/firstleg/controller/EventConsumptionExceptionHandlerTest.java @@ -0,0 +1,22 @@ +package com.baiyue.firstleg.controller; + +import com.baiyue.common.event.EventProtocolException; +import com.baiyue.common.event.EventDependencyNotReadyException; +import org.junit.jupiter.api.Test; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*; + +class EventConsumptionExceptionHandlerTest { + @RestController static class Probe { + @GetMapping("/protocol") void protocol() { throw new EventProtocolException("invalid hash"); } + @GetMapping("/dependency") void dependency() { throw new EventDependencyNotReadyException("source not ready"); } + } + @Test void protocolAndDependencyHaveDistinctWireCodes() throws Exception { + var mvc = MockMvcBuilders.standaloneSetup(new Probe()).setControllerAdvice(new EventConsumptionExceptionHandler()).build(); + mvc.perform(get("/protocol")).andExpect(status().isOk()).andExpect(jsonPath("$.code").value("EVT_PROTOCOL")); + mvc.perform(get("/dependency")).andExpect(status().isOk()).andExpect(jsonPath("$.code").value("EVT_DEPENDENCY")); + } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/client/DirectEventClient.java b/lastleg-service/src/main/java/com/baiyue/lastleg/client/DirectEventClient.java index 4aa290029..4555c23df 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/client/DirectEventClient.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/client/DirectEventClient.java @@ -5,12 +5,17 @@ import com.baiyue.common.event.DeliveryAck; import com.baiyue.common.event.EventEnvelope; import com.baiyue.common.event.EventTypes; import lombok.RequiredArgsConstructor; +import com.baiyue.common.web.EventDeliveryFailure; +import java.util.Set; import org.springframework.stereotype.Component; /** Dispatches each supported business event directly to its owning service. */ @Component @RequiredArgsConstructor public class DirectEventClient { + private static final Set FINANCE_PERMANENT_CODES = Set.of( + "FIN-EVT-001", "FIN-EVT-002", "FIN-EVT-004", "FIN-EVT-007", "FIN-EVT-008", + "FIN-EVT-012", "FIN-EVT-015", "FIN-EVT-016", "FIN-EVT-401"); private final FirstlegEventClient firstleg; private final FinanceBusinessEventClient finance; @@ -20,9 +25,13 @@ public class DirectEventClient { case EventTypes.OVERSEAS_CONTAINER_UNLOADED, EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED, EventTypes.OVERSEAS_DELIVERY_COMPLETED -> firstleg.receive(token, event); - case EventTypes.WALLET_COMMAND_REQUESTED -> finance.wallet(token, event); - case EventTypes.BUSINESS_FEE_SOURCE_CHANGED -> finance.fee(token, event); - default -> throw new IllegalArgumentException("Unsupported outbound event type: " + event.eventType()); + case EventTypes.WALLET_COMMAND_REQUESTED -> checkedFinance(finance.wallet(token, event)); + case EventTypes.BUSINESS_FEE_SOURCE_CHANGED -> checkedFinance(finance.fee(token, event)); + default -> throw new EventDeliveryFailure(true, "Unsupported outbound event type: " + event.eventType()); }; } + private ApiResponse checkedFinance(ApiResponse response) { + EventDeliveryFailure.requireSuccessful(response, FINANCE_PERMANENT_CODES); + return response; + } } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/controller/EventConsumptionExceptionHandler.java b/lastleg-service/src/main/java/com/baiyue/lastleg/controller/EventConsumptionExceptionHandler.java new file mode 100644 index 000000000..15233ca4a --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/controller/EventConsumptionExceptionHandler.java @@ -0,0 +1,21 @@ +package com.baiyue.lastleg.controller; + +import com.baiyue.common.core.api.ApiResponse; +import com.baiyue.common.event.EventProtocolException; +import com.baiyue.common.event.EventDependencyNotReadyException; +import org.slf4j.MDC; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; +import org.springframework.web.bind.annotation.ExceptionHandler; +import org.springframework.web.bind.annotation.RestControllerAdvice; +import java.time.Instant; + +@RestControllerAdvice +@Order(Ordered.HIGHEST_PRECEDENCE) +public class EventConsumptionExceptionHandler { + @ExceptionHandler({EventProtocolException.class, EventDependencyNotReadyException.class}) + public ApiResponse event(IllegalArgumentException failure) { + String code = failure instanceof EventDependencyNotReadyException ? "EVT_DEPENDENCY" : "EVT_PROTOCOL"; + return new ApiResponse<>(code, failure.getMessage(), null, MDC.get("requestId"), MDC.get("traceId"), Instant.now()); + } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java b/lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java index e7312a6f9..0f2043c54 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java @@ -73,7 +73,7 @@ public class LastlegOrderController { } else if (EventTypes.FIRST_LEG_LOADING_PLAN_COMPLETED.equals(event.eventType())) { consumed = consumer.consume(event); } else { - throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Lastleg 不消费该事件类型: " + event.eventType()); + throw new com.baiyue.common.event.EventProtocolException("Lastleg 不消费该事件类型: " + event.eventType()); } return ApiResponse.success(consumed); } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/financefee/service/OverseasFeeOutboxRelay.java b/lastleg-service/src/main/java/com/baiyue/lastleg/financefee/service/OverseasFeeOutboxRelay.java index 571f6c90d..8ab22ae04 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/financefee/service/OverseasFeeOutboxRelay.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/financefee/service/OverseasFeeOutboxRelay.java @@ -1,6 +1,8 @@ package com.baiyue.lastleg.financefee.service; import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.web.EventDeliveryFailure; +import com.fasterxml.jackson.core.JsonProcessingException; import com.baiyue.common.core.retry.BoundedRetryPolicy; import com.baiyue.lastleg.client.DirectEventClient; import com.baiyue.lastleg.entity.LastlegEventOutbox; @@ -46,17 +48,21 @@ public class OverseasFeeOutboxRelay { repository.dead(row.getId(), worker, failures, "自动投递次数已耗尽（含租约超时）"); continue; } + boolean requestStarted = false; try { EventEnvelope envelope = objectMapper.readValue(row.getPayloadJson(), EventEnvelope.class).requireValid(); + requestStarted = true; var response = client.receive(token, envelope); - if (response == null || !response.successful() || response.data() == null) - throw new IllegalStateException("Finance未确认费用事件"); + EventDeliveryFailure.requireSuccessful(response); response.data().requireMatches(envelope); repository.published(row.getId(), worker, LocalDateTime.now(ZoneOffset.UTC)); } catch (Exception exception) { + String error = EventDeliveryFailure.diagnostic(exception, token); + boolean permanent = EventDeliveryFailure.isPermanent(exception) + || (!requestStarted && (exception instanceof JsonProcessingException || exception instanceof IllegalArgumentException)); int retries = failures + 1; - if (policy.exhausted(retries)) repository.dead(row.getId(), worker, retries, exception.getMessage()); - else repository.retry(row.getId(), worker, retries, LocalDateTime.now(ZoneOffset.UTC).plus(policy.delayAfterFailure(retries)), exception.getMessage()); + if (permanent || policy.exhausted(retries)) repository.dead(row.getId(), worker, retries, error); + else repository.retry(row.getId(), worker, retries, LocalDateTime.now(ZoneOffset.UTC).plus(policy.delayAfterFailure(retries)), error); } } } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java index 2cbf52ffb..5749d7013 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java @@ -1,6 +1,8 @@ package com.baiyue.lastleg.inboundplan.pda.service; import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.web.EventDeliveryFailure; +import com.fasterxml.jackson.core.JsonProcessingException; import com.baiyue.common.core.retry.BoundedRetryPolicy; import com.baiyue.common.event.DeliveryAck; import com.baiyue.common.event.EventEnvelopeValidator; @@ -51,29 +53,29 @@ public class InboundUnloadCompletedOutboxRelay { repository.dead(row.getId(), workerId, failures, "自动投递次数已耗尽（含租约超时）"); continue; } + boolean requestStarted = false; try { EventEnvelope event = objectMapper.readValue(row.getPayloadJson(), EventEnvelope.class); EventEnvelopeValidator.requireValid(event); + requestStarted = true; var response = client.receive(internalToken, event); requireAcknowledgement(event, response); repository.published(row.getId(), workerId, LocalDateTime.now()); } catch (Exception exception) { - if (policy.exhausted(failures + 1)) repository.dead(row.getId(), workerId, failures + 1, exception.getMessage()); + String error = EventDeliveryFailure.diagnostic(exception, internalToken); + boolean permanent = EventDeliveryFailure.isPermanent(exception) + || (!requestStarted && (exception instanceof JsonProcessingException || exception instanceof IllegalArgumentException)); + if (permanent || policy.exhausted(failures + 1)) repository.dead(row.getId(), workerId, failures + 1, error); else repository.retry(row.getId(), workerId, LocalDateTime.now().plus(policy.delayAfterFailure(failures + 1)), - exception.getMessage()); - log.warn("海外到国内事件投递失败: eventKey={}, reason={}", row.getEventKey(), exception.getMessage()); + error); + log.warn("海外到国内事件投递失败: eventKey={}, reason={}", row.getEventKey(), error); } } } private void requireAcknowledgement(EventEnvelope event, com.baiyue.common.core.api.ApiResponse response) { - if (response == null) throw new IllegalStateException("消费服务返回空响应"); - if (!response.successful()) { - throw new IllegalStateException("消费服务拒绝海外事件: code=" + response.code() - + ", message=" + response.message() + ", requestId=" + response.requestId()); - } - if (response.data() == null) throw new IllegalStateException("消费服务成功响应缺少确认数据"); + EventDeliveryFailure.requireSuccessful(response); response.data().requireMatches(event); } } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/service/LoadingCompletedConsumer.java b/lastleg-service/src/main/java/com/baiyue/lastleg/service/LoadingCompletedConsumer.java index d3ca0bc7c..12f79d4c1 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/service/LoadingCompletedConsumer.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/service/LoadingCompletedConsumer.java @@ -55,23 +55,23 @@ public class LoadingCompletedConsumer { envelope.requireValid(); if (!EventTypes.FIRST_LEG_LOADING_PLAN_COMPLETED.equals(envelope.eventType()) || (envelope.eventVersion() != 1 && envelope.eventVersion() != 2 && envelope.eventVersion() != 3)) - throw new IllegalArgumentException("不支持的尾程交接事件"); + throw new EventProtocolException("不支持的尾程交接事件"); if (envelope.sourceRegion() != RegionCode.CN || envelope.targetRegion() != RegionCode.OVERSEAS) - throw new IllegalArgumentException("尾程交接事件区域无效"); + throw new EventProtocolException("尾程交接事件区域无效"); if (!"LOADING_PLAN".equals(envelope.aggregateType())) - throw new IllegalArgumentException("尾程交接事件聚合类型无效"); + throw new EventProtocolException("尾程交接事件聚合类型无效"); if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), - envelope.payloadHash())) throw new IllegalArgumentException("payloadHash 与尾程事件负载不一致"); + envelope.payloadHash())) throw new EventProtocolException("payloadHash 与尾程事件负载不一致"); FirstLegLoadingPlanCompletedV1 payload = parsePayload(envelope); String containerNo = ContainerNumbers.requireCanonical(payload.containerNo()); String sourceContainerBookingId = sourceContainerBookingId(envelope, payload); - if(payload.orders()==null||payload.orders().isEmpty()) throw new IllegalArgumentException("事件没有入仓单"); + if(payload.orders()==null||payload.orders().isEmpty()) throw new EventProtocolException("事件没有入仓单"); OwnerSnapshot owner = requireOwnerSnapshot(payload.orders(), envelope.tenantId()); if (envelope.eventVersion() >= 2) validateCartonSkuMappings(payload, envelope.payload()); freezeContainerIdentity(envelope, sourceContainerBookingId, containerNo); LastlegEventInbox existing=inboxMapper.selectById(envelope.eventId()); if(existing!=null) { - if(!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) throw new IllegalArgumentException("事件摘要冲突"); + if(!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) throw new EventProtocolException("事件摘要冲突"); if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费"); return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE); } @@ -84,7 +84,7 @@ public class LoadingCompletedConsumer { LastlegEventInbox raced = inboxMapper.selectById(envelope.eventId()); if (raced == null || raced.getPayloadHash() == null || !raced.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) { - throw new IllegalArgumentException("事件摘要冲突", exception); + throw new EventProtocolException("事件摘要冲突", exception); } if (!"PROCESSED".equals(raced.getStatus())) throw new IllegalStateException("事件尚未完成消费"); return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE); @@ -111,7 +111,7 @@ public class LoadingCompletedConsumer { if (envelope.eventVersion() < 3) return; JsonNode values = envelope.payload().path("loadingEvidence"); if (!values.isArray() || values.size() != 5) { - throw new IllegalArgumentException("V3 尾程交接事件必须包含五类国内装柜照片"); + throw new EventProtocolException("V3 尾程交接事件必须包含五类国内装柜照片"); } Set expected = Set.of("PHOTO_STEP_1", "PHOTO_STEP_2", "PHOTO_STEP_3", "PHOTO_STEP_4", "SEAL"); Set actual = new HashSet<>(); @@ -121,7 +121,7 @@ public class LoadingCompletedConsumer { String fileName = value.path("fileName").asText("").trim(); if (!expected.contains(type) || !actual.add(type) || fileId <= 0 || fileName.isBlank() || fileName.length() > 255) { - throw new IllegalArgumentException("V3 尾程交接事件的国内装柜照片引用无效"); + throw new EventProtocolException("V3 尾程交接事件的国内装柜照片引用无效"); } ContainerLoadingEvidence existing = loadingEvidence.find(sourceContainerBookingId, type); if (existing != null) { @@ -143,7 +143,7 @@ public class LoadingCompletedConsumer { || !Objects.equals(raced.getFileName(), fileName)) throw duplicate; } } - if (!actual.equals(expected)) throw new IllegalArgumentException("V3 尾程交接事件的国内装柜照片类型不完整"); + if (!actual.equals(expected)) throw new EventProtocolException("V3 尾程交接事件的国内装柜照片类型不完整"); } private FirstLegLoadingPlanCompletedV1 parsePayload(EventEnvelope envelope) { @@ -154,7 +154,7 @@ public class LoadingCompletedConsumer { // identity is validated separately from the original immutable payload. return compatibleMapper.treeToValue(envelope.payload(), FirstLegLoadingPlanCompletedV1.class); } - catch(Exception exception) { throw new IllegalArgumentException("尾程事件负载无效",exception); } + catch(Exception exception) { throw new EventProtocolException("尾程事件负载无效",exception); } } private int countOrders(EventEnvelope envelope,String containerNo) { @@ -184,7 +184,7 @@ public class LoadingCompletedConsumer { ? envelope.payload().path("sourceContainerBookingId").asText(null) : payload.containerBookingId(); if (value == null || value.isBlank() || value.trim().length() > 128) { - throw new IllegalArgumentException("事件缺少稳定来源柜预订 ID sourceContainerBookingId"); + throw new EventProtocolException("事件缺少稳定来源柜预订 ID sourceContainerBookingId"); } return value.trim(); } @@ -194,12 +194,12 @@ public class LoadingCompletedConsumer { String customerId = requiredOwnerValue(orders.get(0).customerId(), "customerId"); String customerName = requiredOwnerValue(orders.get(0).customerName(), "customerName"); if (!trustedCustomerId.equals(customerId)) { - throw new IllegalArgumentException("Order customerId does not match the trusted event customer"); + throw new EventProtocolException("Order customerId does not match the trusted event customer"); } for (FirstLegLoadingPlanCompletedV1.Order order : orders) { if (!customerId.equals(requiredOwnerValue(order.customerId(), "customerId")) || !customerName.equals(requiredOwnerValue(order.customerName(), "customerName"))) { - throw new IllegalArgumentException("同一装柜完成事件的货主快照不一致"); + throw new EventProtocolException("同一装柜完成事件的货主快照不一致"); } } return new OwnerSnapshot(customerId, customerName); @@ -207,11 +207,11 @@ public class LoadingCompletedConsumer { private String requiredOwnerValue(String value, String field) { if (value == null || value.isBlank()) { - throw new IllegalArgumentException("尾程事件缺少不可变货主快照 " + field); + throw new EventProtocolException("尾程事件缺少不可变货主快照 " + field); } String normalized = value.trim(); if (normalized.length() > 128) { - throw new IllegalArgumentException("尾程事件货主快照过长 " + field); + throw new EventProtocolException("尾程事件货主快照过长 " + field); } return normalized; } @@ -231,11 +231,11 @@ public class LoadingCompletedConsumer { } int manifestPackages = rawPallets.path(palletIndex).path("manifestPackageCount").asInt(-1); if (palletCartonMarks.size() != manifestPackages) { - throw new IllegalArgumentException("一件代发托盘物理箱数与清单箱数不一致: " + pallet.palletNo()); + throw new EventProtocolException("一件代发托盘物理箱数与清单箱数不一致: " + pallet.palletNo()); } for (String mark : palletCartonMarks) { if (!orderCartonMarks.add(mark)) { - throw new IllegalArgumentException("一件代发物理箱不能跨托盘重复: " + mark); + throw new EventProtocolException("一件代发物理箱不能跨托盘重复: " + mark); } } } @@ -257,7 +257,7 @@ public class LoadingCompletedConsumer { Set orderCartonMarks) { List marks = cartonMarks == null ? List.of() : cartonMarks; if (skuCode == null || skuCode.isBlank()) { - throw new IllegalArgumentException("V2 尾程交接事件缺少 SKU"); + throw new EventProtocolException("V2 尾程交接事件缺少 SKU"); } if (marks.isEmpty()) return; requireSkuCartonMapping(skuCode, loadedPackages, marks, orderCartonMarks); @@ -266,15 +266,15 @@ public class LoadingCompletedConsumer { private void requireSkuCartonMapping(String skuCode, int loadedPackages, List cartonMarks, Set orderCartonMarks) { if (skuCode == null || skuCode.isBlank()) { - throw new IllegalArgumentException("V2 尾程交接事件的箱唛映射缺少 SKU"); + throw new EventProtocolException("V2 尾程交接事件的箱唛映射缺少 SKU"); } List marks = cartonMarks == null ? List.of() : cartonMarks; if (marks.size() != loadedPackages) { - throw new IllegalArgumentException("V2 尾程交接事件的 SKU 箱唛数量必须等于实装箱数: " + skuCode); + throw new EventProtocolException("V2 尾程交接事件的 SKU 箱唛数量必须等于实装箱数: " + skuCode); } for (String mark : marks) { if (mark == null || mark.isBlank() || !orderCartonMarks.add(mark)) { - throw new IllegalArgumentException("V2 尾程交接事件包含空值或重复箱唛: " + skuCode); + throw new EventProtocolException("V2 尾程交接事件包含空值或重复箱唛: " + skuCode); } } } @@ -298,7 +298,7 @@ public class LoadingCompletedConsumer { List dropshipPallets = "ONE_CLICK_DROPSHIP".equals(orderType) ? pallets(source, rawSource) : List.of(); if ("ONE_CLICK_DROPSHIP".equals(orderType) && dropshipPallets.isEmpty()) { - throw new IllegalArgumentException("One-click dropship event has no pallet snapshot"); + throw new EventProtocolException("One-click dropship event has no pallet snapshot"); } order.setId(IdWorker.getId()); order.setOrderNo(orderNo); order.setCustomerId(customerId); order.setOrderType(orderType); order.setDestinationType(destinationType); @@ -346,12 +346,12 @@ public class LoadingCompletedConsumer { for (int palletIndex = 0; palletIndex < pallets.size(); palletIndex++) { String palletNo = pallets.get(palletIndex).palletNo(); if (palletNo == null || palletNo.isBlank() || palletNo.length() > 128 || !palletNos.add(palletNo)) { - throw new IllegalArgumentException("散货来源托盘码为空、重复或过长"); + throw new EventProtocolException("散货来源托盘码为空、重复或过长"); } sourcePallets.insert(IdWorker.getId(), customerId, order.getId(), palletNo.trim(), palletIndex + 1, event.eventId()); } if (!pallets.isEmpty() && order.getPalletCount() > 0 && pallets.size() != order.getPalletCount()) { - throw new IllegalArgumentException("散货来源托盘码数量与托盘数不一致"); + throw new EventProtocolException("散货来源托盘码数量与托盘数不一致"); } } for(var item:items) { @@ -379,7 +379,7 @@ public class LoadingCompletedConsumer { if ("ONE_CLICK_DROPSHIP".equals(orderType)) { long warehouseId; try { warehouseId = Long.parseLong(payload.warehouseId()); } - catch (Exception exception) { throw new IllegalArgumentException("One-click dropship warehouseId is invalid", exception); } + catch (Exception exception) { throw new EventProtocolException("One-click dropship warehouseId is invalid", exception); } dropshipOrders.receive(new DropshipRequests.Receive(customerId, order.getId(), warehouseId, event.eventId() + ":DROPSHIP:" + source.sourceInboundId(), dropshipPallets)); } @@ -390,7 +390,7 @@ public class LoadingCompletedConsumer { try { return LocalDateTime.parse(value.trim()); } catch (java.time.format.DateTimeParseException exception) { - throw new IllegalArgumentException("事件计划到仓时间 estimatedWarehouseArrivalAt 无效", exception); + throw new EventProtocolException("事件计划到仓时间 estimatedWarehouseArrivalAt 无效", exception); } } @@ -447,7 +447,7 @@ public class LoadingCompletedConsumer { String detailId = sku.path("sourceInboundDetailId").asText(null); var item = itemsByDetail.get(detailId); if (item == null || !Objects.equals(item.skuCode(), sku.path("skuCode").asText(null))) { - throw new IllegalArgumentException("一件代发托盘来源明细缺失或与 SKU 不一致"); + throw new EventProtocolException("一件代发托盘来源明细缺失或与 SKU 不一致"); } List marks = marksByDetail.computeIfAbsent(detailId, ignored -> new java.util.ArrayList<>()); sku.path("cartonMarks").forEach(mark -> marks.add(mark.asText())); @@ -455,7 +455,7 @@ public class LoadingCompletedConsumer { } for (var item : itemsByDetail.values()) { if (marksByDetail.getOrDefault(item.sourceInboundDetailId(), List.of()).size() != item.packageCount()) { - throw new IllegalArgumentException("一件代发托盘箱唛与来源明细箱数不一致"); + throw new EventProtocolException("一件代发托盘箱唛与来源明细箱数不一致"); } } return marksByDetail; @@ -466,7 +466,7 @@ public class LoadingCompletedConsumer { Map> byDetail = new java.util.LinkedHashMap<>(); for (var item : source == null ? List.of() : source) { if (item.sourceInboundDetailId() == null || item.sourceInboundDetailId().isBlank()) { - throw new IllegalArgumentException("一件代发来源明细标识缺失"); + throw new EventProtocolException("一件代发来源明细标识缺失"); } byDetail.computeIfAbsent(item.sourceInboundDetailId(), ignored -> new ArrayList<>()).add(item); } @@ -476,7 +476,7 @@ public class LoadingCompletedConsumer { if (fragments.stream().anyMatch(item -> !Objects.equals(first.skuCode(), item.skuCode()) || !Objects.equals(first.destinationWarehouseId(), item.destinationWarehouseId()) || !Objects.equals(first.destinationWarehouseCode(), item.destinationWarehouseCode()))) { - throw new IllegalArgumentException("同一来源明细的尾程事件分片快照不一致"); + throw new EventProtocolException("同一来源明细的尾程事件分片快照不一致"); } List marks = fragments.stream().flatMap(item -> item.cartonMarks().stream()).toList(); result.add(new FirstLegLoadingPlanCompletedV1.Item(first.sourceInboundDetailId(), first.skuCode(), @@ -512,7 +512,7 @@ public class LoadingCompletedConsumer { List marks = sourceMarks == null ? List.of() : sourceMarks; if (marks.isEmpty()) return; if (marks.stream().anyMatch(mark -> mark == null || mark.isBlank()) || marks.stream().distinct().count() != marks.size()) { - throw new IllegalArgumentException("装柜事件的箱唛池不可包含空值或重复箱唛"); + throw new EventProtocolException("装柜事件的箱唛池不可包含空值或重复箱唛"); } for (String mark : marks) { InboundPlanCarton carton = new InboundPlanCarton(); @@ -536,18 +536,18 @@ public class LoadingCompletedConsumer { boolean overseasWarehouseIdentity = hasText(payload.warehouseId()) && Objects.equals(payload.warehouseId(), order.destinationWarehouseId()); if (fbaEvidence && overseasWarehouseIdentity) { - throw new IllegalArgumentException("历史尾程事件目的类型证据冲突: " + order.sourceInboundId()); + throw new EventProtocolException("历史尾程事件目的类型证据冲突: " + order.sourceInboundId()); } if (fbaEvidence) return "FBA"; if (overseasWarehouseIdentity) return "OVERSEAS"; - throw new IllegalArgumentException("历史尾程事件缺少 destinationType 且无法安全推断: " + order.sourceInboundId()); + throw new EventProtocolException("历史尾程事件缺少 destinationType 且无法安全推断: " + order.sourceInboundId()); } private String normalizeDestinationType(String value) { if (!hasText(value)) return null; String normalized = value.trim().toUpperCase(java.util.Locale.ROOT); if (!"FBA".equals(normalized) && !"OVERSEAS".equals(normalized)) { - throw new IllegalArgumentException("尾程事件 destinationType 必须为 FBA 或 OVERSEAS"); + throw new EventProtocolException("尾程事件 destinationType 必须为 FBA 或 OVERSEAS"); } return normalized; } @@ -577,7 +577,7 @@ public class LoadingCompletedConsumer { List marks = new ArrayList<>(left.cartonMarks()); marks.addAll(right.cartonMarks()); Map quantities = new java.util.LinkedHashMap<>(left.cartonQuantities()); right.cartonQuantities().forEach((mark, quantity) -> { - if (quantities.putIfAbsent(mark, quantity) != null) throw new IllegalArgumentException("Duplicate carton quantity: " + mark); + if (quantities.putIfAbsent(mark, quantity) != null) throw new EventProtocolException("Duplicate carton quantity: " + mark); }); return new DropshipRequests.SkuPackages(left.skuCode(), left.productName(), Math.addExact(left.packageCount(), right.packageCount()), @@ -620,25 +620,25 @@ public class LoadingCompletedConsumer { private int positiveManifest(JsonNode source, String field) { int value = source.path(field).asInt(0); - if (value <= 0) throw new IllegalArgumentException("One-click dropship pallet is missing " + field); + if (value <= 0) throw new EventProtocolException("One-click dropship pallet is missing " + field); return value; } private String requiredManifestHash(JsonNode source) { String value = source.path("manifestHash").asText(null); if (value == null || !value.matches("(?i)[0-9a-f]{64}")) { - throw new IllegalArgumentException("One-click dropship pallet manifestHash is invalid"); + throw new EventProtocolException("One-click dropship pallet manifestHash is invalid"); } return value.toLowerCase(java.util.Locale.ROOT); } private Map cartonQuantities(JsonNode sku) { JsonNode node = sku.path("cartonQuantities"); - if (!node.isObject()) throw new IllegalArgumentException("One-click dropship cartonQuantities is required"); + if (!node.isObject()) throw new EventProtocolException("One-click dropship cartonQuantities is required"); Map result = new java.util.LinkedHashMap<>(); node.fields().forEachRemaining(entry -> { if (!entry.getValue().isIntegralNumber() || !entry.getValue().canConvertToLong() || entry.getValue().longValue() <= 0) { - throw new IllegalArgumentException("Invalid carton quantity: " + entry.getKey()); + throw new EventProtocolException("Invalid carton quantity: " + entry.getKey()); } result.put(entry.getKey(), entry.getValue().longValue()); }); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/shipping/service/ContainerShippingNodeConsumer.java b/lastleg-service/src/main/java/com/baiyue/lastleg/shipping/service/ContainerShippingNodeConsumer.java index e140c2cee..0e7b69132 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/shipping/service/ContainerShippingNodeConsumer.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/shipping/service/ContainerShippingNodeConsumer.java @@ -1,6 +1,8 @@ package com.baiyue.lastleg.shipping.service; import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.event.EventProtocolException; +import com.baiyue.common.event.EventDependencyNotReadyException; import com.baiyue.common.event.DeliveryAck; import com.baiyue.common.event.EventPayloadCanonicalizer; import com.baiyue.common.event.EventPayloadHash; @@ -36,7 +38,7 @@ public class ContainerShippingNodeConsumer { LastlegEventInbox existing = inboxes.selectById(envelope.eventId()); if (existing != null) { if (!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) { - throw new IllegalArgumentException("船节点事件摘要冲突"); + throw new EventProtocolException("船节点事件摘要冲突"); } if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费"); return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE); @@ -46,7 +48,7 @@ public class ContainerShippingNodeConsumer { catch (DuplicateKeyException raced) { LastlegEventInbox concurrent = inboxes.selectById(envelope.eventId()); if (concurrent == null || !concurrent.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) { - throw new IllegalArgumentException("船节点事件摘要冲突", raced); + throw new EventProtocolException("船节点事件摘要冲突", raced); } if (!"PROCESSED".equals(concurrent.getStatus())) throw new IllegalStateException("事件尚未完成消费"); return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE); @@ -55,8 +57,9 @@ public class ContainerShippingNodeConsumer { gates.initialize(payload.containerBookingId(), payload.oceanBookingId(), payload.containerNo(), 0, 1); ContainerShippingGate gate = gates.lock(payload.containerBookingId()); - if (gate == null || !payload.containerNo().equals(gate.getContainerNo())) { - throw new IllegalArgumentException("船节点柜号快照与装柜交接不一致"); + if (gate == null) throw new EventDependencyNotReadyException("船节点柜级投影尚未就绪"); + if (!payload.containerNo().equals(gate.getContainerNo())) { + throw new EventProtocolException("船节点柜号快照与装柜交接不一致"); } boolean applied = apply(gate, envelope, payload); gates.updateById(gate); @@ -98,7 +101,7 @@ public class ContainerShippingNodeConsumer { envelope.requireValid(); if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) { - throw new IllegalArgumentException("payloadHash 与船节点事件负载不一致"); + throw new EventProtocolException("payloadHash 与船节点事件负载不一致"); } if (!EventTypes.FIRST_LEG_CONTAINER_SHIPPING_NODE_RECORDED.equals(envelope.eventType()) || envelope.eventVersion() != 1 || envelope.sourceRegion() != RegionCode.CN @@ -106,19 +109,19 @@ public class ContainerShippingNodeConsumer { || !"CONTAINER_BOOKING".equals(envelope.aggregateType()) || envelope.handoverVersion() == null || envelope.handoverVersion() != envelope.aggregateVersion() + 1) { - throw new IllegalArgumentException("船节点交接信封无效"); + throw new EventProtocolException("船节点交接信封无效"); } } private void validateEnvelopePayload(EventEnvelope envelope, FirstLegContainerShippingNodeRecordedV1 payload) { if (!"0".equals(envelope.tenantId()) || !envelope.aggregateId().equals(payload.containerBookingId())) { - throw new IllegalArgumentException("船节点交接平台作用域或聚合标识不一致"); + throw new EventProtocolException("船节点交接平台作用域或聚合标识不一致"); } } private FirstLegContainerShippingNodeRecordedV1 payload(EventEnvelope envelope) { try { return objectMapper.treeToValue(envelope.payload(), FirstLegContainerShippingNodeRecordedV1.class); } - catch (Exception exception) { throw new IllegalArgumentException("船节点交接负载无效", exception); } + catch (Exception exception) { throw new EventProtocolException("船节点交接负载无效", exception); } } private LastlegEventInbox inbox(EventEnvelope envelope) { diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/wallet/WalletCommandCompletedConsumer.java b/lastleg-service/src/main/java/com/baiyue/lastleg/wallet/WalletCommandCompletedConsumer.java index 78a9ad765..ca8858b46 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/wallet/WalletCommandCompletedConsumer.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/wallet/WalletCommandCompletedConsumer.java @@ -14,11 +14,11 @@ import org.springframework.dao.DuplicateKeyException; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; -import org.springframework.web.server.ResponseStatusException; +import com.baiyue.common.event.EventProtocolException; import java.util.List; -import static org.springframework.http.HttpStatus.BAD_REQUEST; + @Service @RequiredArgsConstructor @@ -67,12 +67,12 @@ public class WalletCommandCompletedConsumer { || result.businessType() == null || result.businessType().isBlank() || result.businessNo() == null || result.businessNo().isBlank()) throw bad("资金结果事件字段无效"); return result; - } catch (ResponseStatusException exception) { throw exception; } + } catch (EventProtocolException exception) { throw exception; } catch (Exception exception) { throw bad("资金结果事件载荷无法解析"); } } private String json(Object value) { try { return objectMapper.writeValueAsString(value); } catch (Exception exception) { throw bad("资金结果事件序列化失败"); } } - private static ResponseStatusException bad(String message) { return new ResponseStatusException(BAD_REQUEST, message); } + private static EventProtocolException bad(String message) { return new EventProtocolException(message); } } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/client/DirectEventClientTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/client/DirectEventClientTest.java index 2235aa467..154faaaf0 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/client/DirectEventClientTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/client/DirectEventClientTest.java @@ -27,6 +27,7 @@ class DirectEventClientTest { @Test void feeIsSentOnlyToFinance() { var event = event(EventTypes.BUSINESS_FEE_SOURCE_CHANGED); + when(finance.fee("token", event)).thenReturn(ApiResponse.success(DeliveryAck.of(event, DeliveryAck.Outcome.APPLIED))); client.receive("token", event); verify(finance).fee("token", event); verifyNoMoreInteractions(finance); @@ -45,7 +46,7 @@ class DirectEventClientTest { @Test void unsupportedEventsHaveNoDefaultDestination() { assertThatThrownBy(() -> client.receive("token", event(EventTypes.WALLET_COMMAND_COMPLETED))) - .isInstanceOf(IllegalArgumentException.class).hasMessageContaining("Unsupported outbound"); + .isInstanceOf(com.baiyue.common.web.EventDeliveryFailure.class).hasMessageContaining("Unsupported outbound"); verifyNoInteractions(firstleg, finance); } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/controller/EventConsumptionExceptionHandlerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/controller/EventConsumptionExceptionHandlerTest.java new file mode 100644 index 000000000..1c559321f --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/controller/EventConsumptionExceptionHandlerTest.java @@ -0,0 +1,22 @@ +package com.baiyue.lastleg.controller; + +import com.baiyue.common.event.EventProtocolException; +import com.baiyue.common.event.EventDependencyNotReadyException; +import org.junit.jupiter.api.Test; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*; + +class EventConsumptionExceptionHandlerTest { + @RestController static class Probe { + @GetMapping("/protocol") void protocol() { throw new EventProtocolException("invalid hash"); } + @GetMapping("/dependency") void dependency() { throw new EventDependencyNotReadyException("source not ready"); } + } + @Test void protocolAndDependencyHaveDistinctWireCodes() throws Exception { + var mvc = MockMvcBuilders.standaloneSetup(new Probe()).setControllerAdvice(new EventConsumptionExceptionHandler()).build(); + mvc.perform(get("/protocol")).andExpect(status().isOk()).andExpect(jsonPath("$.code").value("EVT_PROTOCOL")); + mvc.perform(get("/dependency")).andExpect(status().isOk()).andExpect(jsonPath("$.code").value("EVT_DEPENDENCY")); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/controller/LastlegOrderControllerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/controller/LastlegOrderControllerTest.java index 9f311cfbb..d9d71ed41 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/controller/LastlegOrderControllerTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/controller/LastlegOrderControllerTest.java @@ -155,7 +155,7 @@ class LastlegOrderControllerTest { when(event.eventType()).thenReturn(EventTypes.IAM_USER_STATUS_CHANGED); LoadingCompletedConsumer loading = (LoadingCompletedConsumer) ReflectionTestUtils.getField(controller, "consumer"); - assertThrows(ResponseStatusException.class, () -> controller.receive("internal-token", event)); + assertThrows(com.baiyue.common.event.EventProtocolException.class, () -> controller.receive("internal-token", event)); verify(loading, never()).consume(event); } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelayTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelayTest.java index 49d3ee511..d542c183b 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelayTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelayTest.java @@ -75,7 +75,7 @@ class InboundUnloadCompletedOutboxRelayTest { relay.relay(); - verify(repository).retry(eq(9001L), anyString(), any(LocalDateTime.class), eq("consumer unavailable")); + verify(repository).retry(eq(9001L), anyString(), any(LocalDateTime.class), contains("consumer unavailable")); verify(repository, never()).published(anyLong(), anyString(), any()); }