fix(firstleg): expose event bridge relay failures
This commit is contained in:
+30
-4
@@ -1,5 +1,6 @@
|
|||||||
package com.baiyue.firstleg.service;
|
package com.baiyue.firstleg.service;
|
||||||
|
|
||||||
|
import com.baiyue.common.core.api.ApiResponse;
|
||||||
import com.baiyue.common.event.EventEnvelope;
|
import com.baiyue.common.event.EventEnvelope;
|
||||||
import com.baiyue.common.event.EventEnvelopeValidator;
|
import com.baiyue.common.event.EventEnvelopeValidator;
|
||||||
import com.baiyue.firstleg.client.EventBridgeClient;
|
import com.baiyue.firstleg.client.EventBridgeClient;
|
||||||
@@ -38,8 +39,7 @@ public class LoadingCompletedOutboxRelay {
|
|||||||
EventEnvelope event=objectMapper.readValue(row.getPayloadJson(),EventEnvelope.class);
|
EventEnvelope event=objectMapper.readValue(row.getPayloadJson(),EventEnvelope.class);
|
||||||
EventEnvelopeValidator.requireValid(event);
|
EventEnvelopeValidator.requireValid(event);
|
||||||
var response=client.receive(internalToken,event);
|
var response=client.receive(internalToken,event);
|
||||||
if(response==null || !response.successful() || response.data()==null
|
requireAcknowledgement(event, response);
|
||||||
|| !event.eventId().equals(response.data().eventId())) throw new IllegalStateException("事件桥未确认头程跨区事件");
|
|
||||||
repository.published(row.getId(),workerId,LocalDateTime.now());
|
repository.published(row.getId(),workerId,LocalDateTime.now());
|
||||||
} catch(Exception exception) {
|
} catch(Exception exception) {
|
||||||
String error=safeError(exception);
|
String error=safeError(exception);
|
||||||
@@ -50,11 +50,37 @@ public class LoadingCompletedOutboxRelay {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void requireAcknowledgement(EventEnvelope event,
|
||||||
|
ApiResponse<EventBridgeClient.Acknowledgement> 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) {
|
private String safeError(Exception exception) {
|
||||||
String value=exception.getMessage();
|
Throwable diagnostic=diagnosticCause(exception);
|
||||||
if(value==null || value.isBlank()) value=exception.getClass().getSimpleName();
|
String message=diagnostic.getMessage();
|
||||||
|
String value=diagnostic.getClass().getSimpleName()
|
||||||
|
+ ((message==null || message.isBlank())?"":": " + message);
|
||||||
if(internalToken!=null && !internalToken.isBlank()) value=value.replace(internalToken,"***");
|
if(internalToken!=null && !internalToken.isBlank()) value=value.replace(internalToken,"***");
|
||||||
value=value.replaceAll("[\\r\\n\\t]+"," ");
|
value=value.replaceAll("[\\r\\n\\t]+"," ");
|
||||||
return value.length()>500?value.substring(0,500):value;
|
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;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -44,7 +44,7 @@ baiyue:
|
|||||||
jwt-secret: ${JWT_SECRET:change-me-to-at-least-32-bytes-long-before-start}
|
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}
|
internal-service-token: ${INTERNAL_SERVICE_TOKEN:change-me-to-an-internal-token-at-least-32-bytes-long}
|
||||||
lastleg:
|
lastleg:
|
||||||
event-bridge-url: ${LASTLEG_EVENT_BRIDGE_URL:http://127.0.0.1:9312}
|
event-bridge-url: ${LASTLEG_EVENT_BRIDGE_URL:}
|
||||||
clients:
|
clients:
|
||||||
order:
|
order:
|
||||||
url: ${ORDER_SERVICE_URL:}
|
url: ${ORDER_SERVICE_URL:}
|
||||||
|
|||||||
+32
@@ -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]"));
|
||||||
|
}
|
||||||
|
}
|
||||||
+40
-1
@@ -1,6 +1,7 @@
|
|||||||
package com.baiyue.firstleg.service;
|
package com.baiyue.firstleg.service;
|
||||||
|
|
||||||
import com.baiyue.common.core.api.ApiResponse;
|
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.EventEnvelope;
|
||||||
import com.baiyue.common.event.EventPayloadCanonicalizer;
|
import com.baiyue.common.event.EventPayloadCanonicalizer;
|
||||||
import com.baiyue.common.event.EventPayloadHash;
|
import com.baiyue.common.event.EventPayloadHash;
|
||||||
@@ -18,6 +19,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
|
|||||||
import org.mockito.Mock;
|
import org.mockito.Mock;
|
||||||
import org.mockito.junit.jupiter.MockitoExtension;
|
import org.mockito.junit.jupiter.MockitoExtension;
|
||||||
import org.springframework.test.util.ReflectionTestUtils;
|
import org.springframework.test.util.ReflectionTestUtils;
|
||||||
|
import org.springframework.cloud.client.circuitbreaker.NoFallbackAvailableException;
|
||||||
|
|
||||||
import java.time.Instant;
|
import java.time.Instant;
|
||||||
import java.time.LocalDateTime;
|
import java.time.LocalDateTime;
|
||||||
@@ -73,7 +75,44 @@ class LoadingCompletedOutboxRelayTest {
|
|||||||
|
|
||||||
relay.relay();
|
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 {
|
private InboundOutboxEntity row(EventEnvelope event) throws Exception {
|
||||||
|
|||||||
Reference in New Issue
Block a user