fix(event-bridge): restore overseas event delivery
This commit is contained in:
@@ -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;
|
||||
|
||||
+8
-2
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
+16
@@ -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"));
|
||||
|
||||
@@ -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')
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user