java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: enforce canonical container identity across regions

Browse Source
This commit is contained in:
ouyangxq
2026-08-31 12:13:20 +08:00
parent b386f35ab4
commit 942dbace19
78 changed files with 1203 additions and 730 deletions
Download Patch File Download Diff File Expand all files Collapse all files
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/EventInboxService.java
+31 -6
View File
@@ -45,6 +45,11 @@ public class EventInboxService {
if (existing != null) {
return duplicateOrConflict(existing, envelope);
}
String idempotencyKey = idempotencyKey(envelope);
BridgeInbox sameBusinessEvent = inboxStore.findByIdempotencyKey(idempotencyKey);
if (sameBusinessEvent != null) {
return duplicateBusinessEventOrConflict(sameBusinessEvent, envelope);
}
BridgeInbox inbox = toInbox(envelope);
try {
inboxStore.insert(inbox);
@@ -52,13 +57,25 @@ public class EventInboxService {
return new EventReceiveResult(envelope.eventId(), "已接收", inbox.getReceivedAt());
} catch (DuplicateKeyException exception) {
BridgeInbox concurrent = inboxStore.findByEventId(envelope.eventId());
if (concurrent == null) concurrent = inboxStore.findByIdempotencyKey(idempotencyKey);
if (concurrent == null) {
throw exception;
}
return duplicateOrConflict(concurrent, envelope);
return Objects.equals(concurrent.getEventId(), envelope.eventId())
? duplicateOrConflict(concurrent, envelope)
: duplicateBusinessEventOrConflict(concurrent, envelope);
}
}
private EventReceiveResult duplicateBusinessEventOrConflict(BridgeInbox existing, EventEnvelope envelope) {
if (!sameBusinessEvent(existing, envelope)) {
metrics.recordReceive(envelope, EventBridgeMetrics.ReceiveResult.CONFLICT);
throw new IllegalArgumentException("相同eventKey对应的事件负载不一致，拒绝柜号或事实漂移");
}
metrics.recordReceive(envelope, EventBridgeMetrics.ReceiveResult.DUPLICATE);
return new EventReceiveResult(existing.getEventId(), "重复", existing.getReceivedAt());
}
private EventReceiveResult duplicateOrConflict(BridgeInbox existing, EventEnvelope envelope) {
if (!sameEnvelope(existing, envelope)) {
metrics.recordReceive(envelope, EventBridgeMetrics.ReceiveResult.CONFLICT);
@@ -69,6 +86,13 @@ public class EventInboxService {
}
private boolean sameEnvelope(BridgeInbox existing, EventEnvelope envelope) {
return sameBusinessEvent(existing, envelope)
&& Objects.equals(normalizeOccurredAt(existing.getOccurredAt()), normalizeOccurredAt(envelope.occurredAt()))
&& Objects.equals(existing.getTraceId(), envelope.traceId())
&& Objects.equals(existing.getCausationId(), envelope.causationId());
}
private boolean sameBusinessEvent(BridgeInbox existing, EventEnvelope envelope) {
return existing.getPayloadHash() != null
&& existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())
&& Objects.equals(existing.getEventType(), envelope.eventType())
@@ -80,16 +104,13 @@ public class EventInboxService {
&& Objects.equals(existing.getAggregateId(), envelope.aggregateId())
&& Objects.equals(existing.getAggregateVersion(), envelope.aggregateVersion())
&& Objects.equals(existing.getHandoverVersion(), envelope.handoverVersion())
&& Objects.equals(normalizeOccurredAt(existing.getOccurredAt()), normalizeOccurredAt(envelope.occurredAt()))
&& Objects.equals(existing.getTraceId(), envelope.traceId())
&& Objects.equals(existing.getCorrelationId(), envelope.correlationId())
&& Objects.equals(existing.getCausationId(), envelope.causationId());
&& Objects.equals(existing.getCorrelationId(), envelope.correlationId());
}
private BridgeInbox toInbox(EventEnvelope envelope) {
BridgeInbox inbox = new BridgeInbox();
inbox.setEventId(envelope.eventId());
inbox.setIdempotencyKey(envelope.sourceRegion() + ":" + envelope.eventId());
inbox.setIdempotencyKey(idempotencyKey(envelope));
inbox.setEventType(envelope.eventType());
inbox.setEventVersion(envelope.eventVersion());
inbox.setSourceRegion(envelope.sourceRegion().name());
@@ -117,4 +138,8 @@ public class EventInboxService {
private static Instant normalizeOccurredAt(Instant value) {
return value == null ? null : value.truncatedTo(ChronoUnit.MILLIS);
}
private static String idempotencyKey(EventEnvelope envelope) {
return envelope.sourceRegion() + ":" + envelope.correlationId();
}
}
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/InboxStore.java
+1
View File
@@ -7,5 +7,6 @@ import com.baiyue.eventbridge.domain.BridgeInbox;
*/
public interface InboxStore {
BridgeInbox findByEventId(String eventId);
BridgeInbox findByIdempotencyKey(String idempotencyKey);
void insert(BridgeInbox inbox);
}
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/MybatisInboxStore.java
+7
View File
@@ -4,6 +4,7 @@ import com.baiyue.eventbridge.domain.BridgeInbox;
import com.baiyue.eventbridge.mapper.BridgeInboxMapper;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Repository;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
/**
* 基于 MyBatis-Plus 的入站幂等存储。
@@ -17,6 +18,12 @@ public class MybatisInboxStore implements InboxStore {
return mapper.selectById(eventId);
}
@Override
public BridgeInbox findByIdempotencyKey(String idempotencyKey) {
return mapper.selectOne(new LambdaQueryWrapper<BridgeInbox>()
.eq(BridgeInbox::getIdempotencyKey, idempotencyKey));
}
@Override
public void insert(BridgeInbox inbox) {
mapper.insert(inbox);
event-bridge-service/src/main/resources/db/event-bridge.sql
+1 -1
View File
@@ -34,7 +34,7 @@ CREATE TABLE IF NOT EXISTS bridge_outbox (
CREATE TABLE IF NOT EXISTS bridge_inbox (
event_id VARCHAR(128) NOT NULL COMMENT '事件全局标识',
idempotency_key VARCHAR(191) NOT NULL COMMENT '接收幂等键',
idempotency_key VARCHAR(191) NOT NULL COMMENT '来源区域与业务eventKey组成的接收幂等键',
event_type VARCHAR(64) NOT NULL COMMENT '事件类型',
event_version INT NOT NULL COMMENT '事件契约版本',
source_region VARCHAR(32) NOT NULL COMMENT '来源区域',
event-bridge-service/src/test/java/com/baiyue/eventbridge/service/EventInboxServiceTest.java
+26
View File
@@ -106,6 +106,32 @@ class EventInboxServiceTest {
.hasMessageContaining("事件信封");
}
@Test
void 相同业务eventKey的新eventId只允许同负载重放并拒绝柜号漂移() {
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
BridgeInbox existing = existing(envelope, envelope.payloadHash());
when(store.findByEventId(envelope.eventId())).thenReturn(null);
when(store.findByIdempotencyKey("CN:flow-1")).thenReturn(existing);
EventEnvelope replay = new EventEnvelope("550e8400-e29b-41d4-a716-446655440001",
envelope.eventType(), envelope.eventVersion(), Instant.parse("2026-07-29T09:00:00Z"),
envelope.sourceRegion(), envelope.targetRegion(), envelope.tenantId(), envelope.aggregateType(),
envelope.aggregateId(), envelope.aggregateVersion(), envelope.handoverVersion(), "trace-2",
envelope.correlationId(), null, envelope.payload(), envelope.payloadHash());
assertThat(service.receive(replay).result()).isEqualTo("重复");
com.fasterxml.jackson.databind.node.ObjectNode driftPayload =
(com.fasterxml.jackson.databind.node.ObjectNode) envelope.payload().deepCopy();
driftPayload.put("containerNo", "DIFFERENT");
String driftHash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(driftPayload));
EventEnvelope drift = new EventEnvelope("550e8400-e29b-41d4-a716-446655440002",
envelope.eventType(), envelope.eventVersion(), envelope.occurredAt(), envelope.sourceRegion(),
envelope.targetRegion(), envelope.tenantId(), envelope.aggregateType(), envelope.aggregateId(),
envelope.aggregateVersion(), envelope.handoverVersion(), "trace-3", envelope.correlationId(),
null, driftPayload, driftHash);
assertThatThrownBy(() -> service.receive(drift))
.hasMessageContaining("eventKey").hasMessageContaining("漂移");
}
@Test
void 并发插入冲突后按既有记录判定为重复() {
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));