From 15421db4215f57d5eb908e3ab68265d570f4f96b Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Wed, 26 Aug 2026 11:06:24 +0800 Subject: [PATCH] fix(event-bridge): restore overseas event delivery --- deploy/sql/海外区域统一增量-MySQL8.sql | 107 ++++++++++++++++++ .../service/EventInboxService.java | 10 +- .../service/EventInboxServiceTest.java | 16 +++ scripts/overseas-deployment-contract.test.mjs | 18 +++ 4 files changed, 149 insertions(+), 2 deletions(-) diff --git a/deploy/sql/海外区域统一增量-MySQL8.sql b/deploy/sql/海外区域统一增量-MySQL8.sql index 882d2a042..7504a0f81 100644 --- a/deploy/sql/海外区域统一增量-MySQL8.sql +++ b/deploy/sql/海外区域统一增量-MySQL8.sql @@ -1,6 +1,113 @@ -- 海外区域固定增量入口。空库先执行 lastleg-service/src/main/resources/db/lastleg.sql 基线； +-- 本入口同时初始化海外 Event Bridge 独立库，避免国内/海外实例误共用 Inbox。 -- 本文件只由人工审核执行，服务部署默认不自动执行 SQL。 SET NAMES utf8mb4; + +CREATE DATABASE IF NOT EXISTS baiyue_event_bridge_overseas + DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; +USE baiyue_event_bridge_overseas; + +CREATE TABLE IF NOT EXISTS bridge_outbox ( + event_id VARCHAR(128) NOT NULL COMMENT '事件全局标识', + idempotency_key VARCHAR(191) NOT NULL COMMENT '发送幂等键', + event_type VARCHAR(64) NOT NULL COMMENT '事件类型', + event_version INT NOT NULL COMMENT '事件契约版本', + source_region VARCHAR(32) NOT NULL COMMENT '来源区域', + target_region VARCHAR(32) NOT NULL COMMENT '目标区域', + customer_id VARCHAR(128) NOT NULL COMMENT '客户标识', + aggregate_type VARCHAR(64) NOT NULL COMMENT '聚合类型', + aggregate_id VARCHAR(128) NOT NULL COMMENT '聚合标识', + aggregate_version BIGINT NOT NULL COMMENT '聚合版本', + handover_version BIGINT NULL COMMENT '跨区交接版本', + trace_id VARCHAR(128) NOT NULL COMMENT '链路标识', + correlation_id VARCHAR(128) NOT NULL COMMENT '业务关联标识', + causation_id VARCHAR(128) NULL COMMENT '上游事件标识', + occurred_at DATETIME(3) NOT NULL COMMENT '业务事实发生时间', + payload_hash VARCHAR(71) NOT NULL COMMENT '规范化负载摘要', + payload_json JSON NOT NULL COMMENT '事件负载', + status VARCHAR(32) NOT NULL DEFAULT 'PENDING' COMMENT '投递状态', + retry_count INT NOT NULL DEFAULT 0 COMMENT '已重试次数', + next_retry_at DATETIME(3) NULL COMMENT '下次重试时间', + lease_owner VARCHAR(128) NULL COMMENT '租约持有者', + lease_until DATETIME(3) NULL COMMENT '租约截止时间', + last_error VARCHAR(1000) NULL COMMENT '最后错误', + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', + delivered_at DATETIME(3) NULL COMMENT '投递成功时间', + PRIMARY KEY (event_id), + UNIQUE KEY uk_bridge_outbox_idempotency (idempotency_key), + KEY idx_bridge_outbox_schedule (target_region,status,next_retry_at,lease_until) +) ENGINE=InnoDB COMMENT='海外跨区域事件待投递箱'; + +CREATE TABLE IF NOT EXISTS bridge_inbox ( + event_id VARCHAR(128) NOT NULL COMMENT '事件全局标识', + idempotency_key VARCHAR(191) NOT NULL COMMENT '接收幂等键', + event_type VARCHAR(64) NOT NULL COMMENT '事件类型', + event_version INT NOT NULL COMMENT '事件契约版本', + source_region VARCHAR(32) NOT NULL COMMENT '来源区域', + target_region VARCHAR(32) NOT NULL COMMENT '目标区域', + customer_id VARCHAR(128) NOT NULL COMMENT '客户标识', + aggregate_type VARCHAR(64) NOT NULL COMMENT '聚合类型', + aggregate_id VARCHAR(128) NOT NULL COMMENT '聚合标识', + aggregate_version BIGINT NOT NULL COMMENT '聚合版本', + handover_version BIGINT NULL COMMENT '跨区交接版本', + trace_id VARCHAR(128) NOT NULL COMMENT '链路标识', + correlation_id VARCHAR(128) NOT NULL COMMENT '业务关联标识', + causation_id VARCHAR(128) NULL COMMENT '上游事件标识', + occurred_at DATETIME(3) NOT NULL COMMENT '业务事实发生时间', + payload_hash VARCHAR(71) NOT NULL COMMENT '规范化负载摘要', + payload_json JSON NOT NULL COMMENT '事件负载', + status VARCHAR(32) NOT NULL DEFAULT 'RECEIVED' COMMENT '消费状态', + lease_owner VARCHAR(128) NULL COMMENT '消费租约持有者', + lease_until DATETIME(3) NULL COMMENT '消费租约截止时间', + retry_count INT NOT NULL DEFAULT 0 COMMENT '消费重试次数', + next_retry_at DATETIME(3) NULL COMMENT '下次消费时间', + last_error VARCHAR(1000) NULL COMMENT '最后错误', + received_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '接收时间', + processed_at DATETIME(3) NULL COMMENT '消费完成时间', + PRIMARY KEY (event_id), + UNIQUE KEY uk_bridge_inbox_idempotency (idempotency_key), + KEY idx_bridge_inbox_consume (status,next_retry_at,lease_until) +) ENGINE=InnoDB COMMENT='海外跨区域事件已接收箱'; + +CREATE TABLE IF NOT EXISTS bridge_delivery_attempt ( + id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键', + event_id VARCHAR(128) NOT NULL COMMENT '事件全局标识', + source_region VARCHAR(32) NOT NULL COMMENT '来源区域', + target_region VARCHAR(32) NOT NULL COMMENT '目标区域', + attempt_no INT NOT NULL COMMENT '投递序号', + status VARCHAR(32) NOT NULL COMMENT '投递结果', + response_code VARCHAR(64) NULL COMMENT '对端响应码', + error_message VARCHAR(1000) NULL COMMENT '错误信息', + started_at DATETIME(3) NOT NULL COMMENT '开始时间', + finished_at DATETIME(3) NULL COMMENT '完成时间', + PRIMARY KEY (id), + UNIQUE KEY uk_bridge_attempt (event_id,attempt_no), + KEY idx_bridge_attempt_event (event_id,started_at) +) ENGINE=InnoDB COMMENT='海外跨区域事件投递尝试记录'; + +DROP PROCEDURE IF EXISTS assert_overseas_event_bridge_schema; +DELIMITER $$ +CREATE PROCEDURE assert_overseas_event_bridge_schema() +BEGIN + IF (SELECT COUNT(*) FROM information_schema.tables + WHERE table_schema='baiyue_event_bridge_overseas' + AND table_name IN ('bridge_outbox','bridge_inbox','bridge_delivery_attempt')) <> 3 THEN + SIGNAL SQLSTATE '45000' + SET MESSAGE_TEXT='海外区域初始化失败：Event Bridge 三张表未完整创建'; + END IF; + IF (SELECT COUNT(DISTINCT CONCAT(table_name, ':', index_name)) FROM information_schema.statistics + WHERE table_schema='baiyue_event_bridge_overseas' + AND ((table_name='bridge_outbox' AND index_name IN ('PRIMARY','uk_bridge_outbox_idempotency','idx_bridge_outbox_schedule')) + OR (table_name='bridge_inbox' AND index_name IN ('PRIMARY','uk_bridge_inbox_idempotency','idx_bridge_inbox_consume')) + OR (table_name='bridge_delivery_attempt' AND index_name IN ('PRIMARY','uk_bridge_attempt','idx_bridge_attempt_event')))) <> 9 THEN + SIGNAL SQLSTATE '45000' + SET MESSAGE_TEXT='海外区域初始化失败：Event Bridge 索引未完整创建'; + END IF; +END$$ +DELIMITER ; +CALL assert_overseas_event_bridge_schema(); +DROP PROCEDURE IF EXISTS assert_overseas_event_bridge_schema; + CREATE DATABASE IF NOT EXISTS baiyue_lastleg DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE baiyue_lastleg; diff --git a/event-bridge-service/src/main/java/com/baiyue/eventbridge/service/EventInboxService.java b/event-bridge-service/src/main/java/com/baiyue/eventbridge/service/EventInboxService.java index f4adf8f3d..f904c7984 100644 --- a/event-bridge-service/src/main/java/com/baiyue/eventbridge/service/EventInboxService.java +++ b/event-bridge-service/src/main/java/com/baiyue/eventbridge/service/EventInboxService.java @@ -14,7 +14,9 @@ import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import jakarta.annotation.Resource; +import java.time.Instant; import java.time.LocalDateTime; +import java.time.temporal.ChronoUnit; import java.util.Objects; /** @@ -78,7 +80,7 @@ public class EventInboxService { && Objects.equals(existing.getAggregateId(), envelope.aggregateId()) && Objects.equals(existing.getAggregateVersion(), envelope.aggregateVersion()) && Objects.equals(existing.getHandoverVersion(), envelope.handoverVersion()) - && Objects.equals(existing.getOccurredAt(), envelope.occurredAt()) + && 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()); @@ -100,7 +102,7 @@ public class EventInboxService { inbox.setTraceId(envelope.traceId()); inbox.setCorrelationId(envelope.correlationId()); inbox.setCausationId(envelope.causationId()); - inbox.setOccurredAt(envelope.occurredAt()); + inbox.setOccurredAt(normalizeOccurredAt(envelope.occurredAt())); inbox.setPayloadHash(envelope.payloadHash().toLowerCase(java.util.Locale.ROOT)); try { inbox.setPayloadJson(objectMapper.writeValueAsString(envelope.payload())); @@ -111,4 +113,8 @@ public class EventInboxService { inbox.setReceivedAt(LocalDateTime.now()); return inbox; } + + private static Instant normalizeOccurredAt(Instant value) { + return value == null ? null : value.truncatedTo(ChronoUnit.MILLIS); + } } diff --git a/event-bridge-service/src/test/java/com/baiyue/eventbridge/service/EventInboxServiceTest.java b/event-bridge-service/src/test/java/com/baiyue/eventbridge/service/EventInboxServiceTest.java index c8ee93232..6789ed520 100644 --- a/event-bridge-service/src/test/java/com/baiyue/eventbridge/service/EventInboxServiceTest.java +++ b/event-bridge-service/src/test/java/com/baiyue/eventbridge/service/EventInboxServiceTest.java @@ -65,6 +65,22 @@ class EventInboxServiceTest { assertThat(receiveCount("DUPLICATE")).isEqualTo(1.0); } + @Test + void 数据库毫秒精度截断后相同事件仍判定为重复() { + EventEnvelope source = envelope("sha256:placeholder"); + EventEnvelope precise = new EventEnvelope(source.eventId(), source.eventType(), source.eventVersion(), + Instant.parse("2026-07-29T08:30:00.123456789Z"), source.sourceRegion(), source.targetRegion(), + source.tenantId(), source.aggregateType(), source.aggregateId(), source.aggregateVersion(), + source.handoverVersion(), source.traceId(), source.correlationId(), source.causationId(), + source.payload(), source.payloadHash()); + EventEnvelope envelope = withHash(precise); + BridgeInbox existing = existing(envelope, envelope.payloadHash()); + existing.setOccurredAt(Instant.parse("2026-07-29T08:30:00.123Z")); + when(store.findByEventId(envelope.eventId())).thenReturn(existing); + + assertThat(service.receive(envelope).result()).isEqualTo("重复"); + } + @Test void 相同事件但摘要不同时拒绝覆盖() { EventEnvelope envelope = withHash(envelope("sha256:placeholder")); diff --git a/scripts/overseas-deployment-contract.test.mjs b/scripts/overseas-deployment-contract.test.mjs index d5c3efac4..72155a2ea 100644 --- a/scripts/overseas-deployment-contract.test.mjs +++ b/scripts/overseas-deployment-contract.test.mjs @@ -131,3 +131,21 @@ test('domestic event bridge SQL converges legacy tenant columns to the customer assert.match(incremental, /bridge_inbox_customer_column = 0 AND @bridge_inbox_tenant_column = 1/) assert.match(incremental, /bridge_outbox_customer_column = 0 AND @bridge_outbox_tenant_column = 1/) }) + +test('overseas unified SQL initializes the isolated Event Bridge database before services start', async () => { + const overseas = await read('deploy/sql/海外区域统一增量-MySQL8.sql') + assert.match(overseas, /CREATE DATABASE IF NOT EXISTS baiyue_event_bridge_overseas/) + assert.match(overseas, /USE baiyue_event_bridge_overseas;/) + for (const table of ['bridge_outbox', 'bridge_inbox', 'bridge_delivery_attempt']) { + assert.match(overseas, new RegExp(`CREATE TABLE IF NOT EXISTS ${table}`)) + } + for (const index of [ + 'uk_bridge_outbox_idempotency', 'idx_bridge_outbox_schedule', + 'uk_bridge_inbox_idempotency', 'idx_bridge_inbox_consume', + 'uk_bridge_attempt', 'idx_bridge_attempt_event' + ]) assert.match(overseas, new RegExp(index)) + assert.match(overseas, /COUNT\(DISTINCT CONCAT\(table_name, ':', index_name\)\)/) + assert.match(overseas, /CALL assert_overseas_event_bridge_schema\(\);/) + assert.ok(overseas.indexOf('USE baiyue_event_bridge_overseas;') < overseas.indexOf('USE baiyue_lastleg;'), + 'overseas Event Bridge schema must be initialized before Lastleg tables') +})