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 c635fbe76..f01597951 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelay.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelay.java @@ -1,5 +1,6 @@ package com.baiyue.firstleg.service; +import com.baiyue.common.core.api.ApiResponse; import com.baiyue.common.event.EventEnvelope; import com.baiyue.common.event.EventEnvelopeValidator; import com.baiyue.firstleg.client.EventBridgeClient; @@ -38,8 +39,7 @@ public class LoadingCompletedOutboxRelay { EventEnvelope event=objectMapper.readValue(row.getPayloadJson(),EventEnvelope.class); EventEnvelopeValidator.requireValid(event); var response=client.receive(internalToken,event); - if(response==null || !response.successful() || response.data()==null - || !event.eventId().equals(response.data().eventId())) throw new IllegalStateException("事件桥未确认头程跨区事件"); + requireAcknowledgement(event, response); repository.published(row.getId(),workerId,LocalDateTime.now()); } catch(Exception exception) { String error=safeError(exception); @@ -50,11 +50,37 @@ public class LoadingCompletedOutboxRelay { } } + private void requireAcknowledgement(EventEnvelope event, + ApiResponse response) { + if(response==null) throw new IllegalStateException("事件桥未返回响应"); + if(!response.successful()) { + throw new IllegalStateException("事件桥拒绝头程跨区事件: code=" + response.code() + + ", message=" + response.message() + ", requestId=" + response.requestId() + + ", traceId=" + response.traceId()); + } + if(response.data()==null || !event.eventId().equals(response.data().eventId())) { + throw new IllegalStateException("事件桥 ACK 不匹配: expectedEventId=" + event.eventId() + + ", actualEventId=" + (response.data()==null?null:response.data().eventId())); + } + } + private String safeError(Exception exception) { - String value=exception.getMessage(); - if(value==null || value.isBlank()) value=exception.getClass().getSimpleName(); + 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/resources/application.yml b/firstleg/src/main/resources/application.yml index ed6994938..1a5342f3a 100644 --- a/firstleg/src/main/resources/application.yml +++ b/firstleg/src/main/resources/application.yml @@ -44,7 +44,7 @@ baiyue: jwt-secret: ${JWT_SECRET:change-me-to-at-least-32-bytes-long-before-start} internal-service-token: ${INTERNAL_SERVICE_TOKEN:change-me-to-an-internal-token-at-least-32-bytes-long} lastleg: - event-bridge-url: ${LASTLEG_EVENT_BRIDGE_URL:http://127.0.0.1:9312} + event-bridge-url: ${LASTLEG_EVENT_BRIDGE_URL:} clients: order: url: ${ORDER_SERVICE_URL:} diff --git a/firstleg/src/test/java/com/baiyue/firstleg/client/EventBridgeClientDeploymentContractTest.java b/firstleg/src/test/java/com/baiyue/firstleg/client/EventBridgeClientDeploymentContractTest.java new file mode 100644 index 000000000..b22423e5f --- /dev/null +++ b/firstleg/src/test/java/com/baiyue/firstleg/client/EventBridgeClientDeploymentContractTest.java @@ -0,0 +1,32 @@ +package com.baiyue.firstleg.client; + +import org.junit.jupiter.api.Test; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class EventBridgeClientDeploymentContractTest { + private static final Path REPOSITORY_ROOT = Path.of("..").toAbsolutePath().normalize(); + + @Test + void firstlegBridgeTargetComesFromEnvironmentWithoutLoopbackFallback() throws Exception { + String application = Files.readString(REPOSITORY_ROOT.resolve("firstleg/src/main/resources/application.yml")); + String nacos = Files.readString(REPOSITORY_ROOT.resolve("config/nacos/cn/firstleg-service.yml")); + + assertTrue(application.contains("event-bridge-url: ${LASTLEG_EVENT_BRIDGE_URL:}")); + assertFalse(application.contains("LASTLEG_EVENT_BRIDGE_URL:http://127.0.0.1")); + assertTrue(nacos.contains("event-bridge-url: ${LASTLEG_EVENT_BRIDGE_URL}")); + } + + @Test + void cnComposeForwardsTargetAndPublishesStableNetworkAlias() throws Exception { + String compose = Files.readString(REPOSITORY_ROOT.resolve("deploy/docker-compose.yml")); + + assertTrue(compose.contains("LASTLEG_EVENT_BRIDGE_URL: ${LASTLEG_EVENT_BRIDGE_URL}")); + assertTrue(compose.contains("SERVER_PORT: 9312")); + assertTrue(compose.contains("aliases: [event-bridge-cn]")); + } +} diff --git a/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelayTest.java b/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelayTest.java index abd8ea172..3970763f6 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelayTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelayTest.java @@ -1,6 +1,7 @@ package com.baiyue.firstleg.service; import com.baiyue.common.core.api.ApiResponse; +import com.baiyue.common.core.api.ResultCode; import com.baiyue.common.event.EventEnvelope; import com.baiyue.common.event.EventPayloadCanonicalizer; import com.baiyue.common.event.EventPayloadHash; @@ -18,6 +19,7 @@ import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.cloud.client.circuitbreaker.NoFallbackAvailableException; import java.time.Instant; import java.time.LocalDateTime; @@ -73,7 +75,44 @@ class LoadingCompletedOutboxRelayTest { relay.relay(); - verify(repository).retry(eq(9001L), anyString(), any(LocalDateTime.class), eq("failed *** remote")); + verify(repository).retry(eq(9001L), anyString(), any(LocalDateTime.class), + eq("IllegalStateException: failed *** remote")); + } + + @Test + void persistsRootCauseWhenCircuitBreakerWrapsBridgeFailure() throws Exception { + EventEnvelope event = event(); + InboundOutboxEntity row = row(event); + when(repository.pending(eq(50), any(), any())).thenReturn(List.of(row)); + when(repository.claim(eq(9001L), anyString(), any(), any())).thenReturn(true); + when(repository.claimed(eq(9001L), anyString())).thenReturn(row); + when(client.receive(eq("internal-token"), any(EventEnvelope.class))).thenThrow( + new NoFallbackAvailableException("No fallback available.", + new IllegalStateException("Connection refused: event-bridge-cn:9312"))); + + relay.relay(); + + verify(repository).retry(eq(9001L), anyString(), any(LocalDateTime.class), + eq("IllegalStateException: Connection refused: event-bridge-cn:9312")); + verify(repository, never()).published(anyLong(), anyString(), any()); + } + + @Test + void persistsDownstreamCodeAndRequestEvidenceWhenBridgeRejectsEvent() throws Exception { + EventEnvelope event = event(); + InboundOutboxEntity row = row(event); + when(repository.pending(eq(50), any(), any())).thenReturn(List.of(row)); + when(repository.claim(eq(9001L), anyString(), any(), any())).thenReturn(true); + when(repository.claimed(eq(9001L), anyString())).thenReturn(row); + when(client.receive(eq("internal-token"), any(EventEnvelope.class))).thenReturn(ApiResponse.failure( + ResultCode.SERVICE_UNAVAILABLE, "bridge inbox unavailable", "request-123", "trace-456")); + + relay.relay(); + + verify(repository).retry(eq(9001L), anyString(), any(LocalDateTime.class), eq( + "IllegalStateException: 事件桥拒绝头程跨区事件: code=S0503, message=bridge inbox unavailable, " + + "requestId=request-123, traceId=trace-456")); + verify(repository, never()).published(anyLong(), anyString(), any()); } private InboundOutboxEntity row(EventEnvelope event) throws Exception {