MySQL 到 Elasticsearch 的一致性：Outbox、Revision、幂等写入与 Repair Job

MySQL
→ 保存页面、权限、目录、关系、任务、版本等业务事实

Elasticsearch
→ 保存面向 BM25、向量检索、过滤和聚合的搜索投影

MySQL Source of Truth
+
Transactional Outbox
+
单调 Source Revision
+
确定性全量投影
+
Elasticsearch External Version
+
Tombstone
+
重试与 Dead Letter
+
Repair Job
+
物理索引 v1/v2 与稳定 Alias

{
  "id": "page_001",
  "title": "齐夏人物分析",
  "revision": 42,
  "relations": {
    "RELATED_CHARACTER": [
      {
        "entity_id": "character_xxx",
        "entity_type": "character",
        "title": "齐夏"
      }
    ]
  }
}

{
  "id": "page_001",
  "title": "齐夏人物分析",
  "relations": {
    "RELATED_CHARACTER": [
      "character:character_xxx"
    ]
  },
  "relation_target_ids": [
    "character:character_xxx"
  ],
  "source_revision": 42,
  "projection_schema_version": 2,
  "is_deleted": false
}

ES 文档必须能够证明：
它来自哪个 MySQL 实体、
哪个业务 revision、
哪一版投影规则。

业务行与 Outbox
在同一个 MySQL 本地事务中提交。

func UpdatePage(ctx context.Context, page Page) error {
    if err := mysql.UpdatePage(ctx, page); err != nil {
        return err
    }

    if err := elastic.IndexPage(ctx, page); err != nil {
        return err
    }

    return nil
}

客户端收到 timeout
不等于
Elasticsearch 一定没有完成写入。

业务事务原子
+
索引最终一致
+
失败可恢复
+
延迟可观测
+
整库可重建

CREATE TABLE wiki_page (
    id              VARCHAR(64) NOT NULL,
    title           VARCHAR(512) NOT NULL,
    content         MEDIUMTEXT NOT NULL,
    relations_json  JSON NOT NULL,
    revision        BIGINT UNSIGNED NOT NULL,
    deleted_at      DATETIME(6) NULL,
    updated_at      DATETIME(6) NOT NULL,
    PRIMARY KEY (id)
);

创建        revision = 1
第一次修改  revision = 2
第二次修改  revision = 3
删除        revision = 4
恢复        revision = 5

CREATE TABLE search_outbox (
    id                  BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    event_id            CHAR(36) NOT NULL,
    aggregate_type      VARCHAR(64) NOT NULL,
    aggregate_id        VARCHAR(128) NOT NULL,
    operation           VARCHAR(16) NOT NULL,
    source_revision     BIGINT UNSIGNED NOT NULL,

    status              VARCHAR(16) NOT NULL DEFAULT 'PENDING',
    available_at        DATETIME(6) NOT NULL,
    attempts            INT UNSIGNED NOT NULL DEFAULT 0,
    locked_by           VARCHAR(128) NULL,
    locked_at           DATETIME(6) NULL,
    last_error          TEXT NULL,

    created_at          DATETIME(6) NOT NULL,
    updated_at          DATETIME(6) NOT NULL,

    PRIMARY KEY (id),
    UNIQUE KEY uk_search_outbox_event (event_id),
    KEY idx_search_outbox_poll (status, available_at, id),
    KEY idx_search_outbox_entity (
        aggregate_type,
        aggregate_id,
        source_revision
    )
);

UPSERT
TOMBSTONE
REBUILD

func UpdatePage(
    ctx context.Context,
    cmd UpdatePageCommand,
) error {
    return mysql.WithTx(ctx, func(tx *sql.Tx) error {
        page, err := pageRepo.GetForUpdate(ctx, tx, cmd.PageID)
        if err != nil {
            return err
        }

        page.Apply(cmd)
        page.Revision++

        if err := pageRepo.Save(ctx, tx, page); err != nil {
            return err
        }

        event := SearchOutboxEvent{
            EventID:        uuid.NewString(),
            AggregateType:  "wiki_page",
            AggregateID:    page.ID,
            Operation:      "UPSERT",
            SourceRevision: page.Revision,
            AvailableAt:    time.Now(),
        }

        return outboxRepo.Insert(ctx, tx, event)
    })
}

业务行和 Outbox 都提交
或
业务行和 Outbox 都回滚

SELECT *
FROM search_outbox
WHERE status = 'PENDING'
  AND available_at <= NOW(6)
ORDER BY id
LIMIT 100
FOR UPDATE SKIP LOCKED;

应用只向 Outbox 追加事件
→ Debezium 读取 MySQL Binlog
→ Outbox Event Router 转换事件
→ Kafka
→ ES Projection Consumer

唯一事件 ID
聚合类型
聚合 ID
事件类型
Payload

{
  "event_id": "evt-001",
  "aggregate_type": "wiki_page",
  "aggregate_id": "page_001",
  "operation": "UPSERT",
  "source_revision": 42
}

{
  "event_id": "evt-001",
  "aggregate_id": "page_001",
  "source_revision": 42,
  "operation": "UPSERT",
  "payload": {
    "title": "齐夏人物分析",
    "relation_target_ids": [
      "character:character_xxx"
    ]
  }
}

Thin Event
+
业务行软删除或独立 Tombstone
+
Worker 读取当前最新事实

这个实体至少发生过一次需要重新投影的变化。

func BuildWikiProjection(page Page) WikiDocument {
    relations, targets := ProjectRelations(page.Relations)

    return WikiDocument{
        ID:                      page.ID,
        Title:                   page.Title,
        Content:                 page.Content,
        Relations:               relations,
        RelationTargetIDs:       targets,
        SourceRevision:          page.Revision,
        ProjectionSchemaVersion: 2,
        IsDeleted:               page.DeletedAt.Valid,
    }
}

相同业务事实
+
相同投影代码
=
完全相同的搜索文档

revision 41 → Worker B 处理较慢
revision 42 → Worker A 处理较快

42 先写入 ES
41 后写入 ES

{
  "mappings": {
    "dynamic": "strict",
    "properties": {
      "id": {
        "type": "keyword"
      },
      "source_revision": {
        "type": "long"
      },
      "projection_schema_version": {
        "type": "integer"
      },
      "projection_hash": {
        "type": "keyword"
      },
      "is_deleted": {
        "type": "boolean"
      }
    }
  }
}

PUT kb_gateway_wiki/_doc/page_001
    ?version=42
    &version_type=external
    &require_alias=true

{
  "id": "page_001",
  "title": "齐夏人物分析",
  "source_revision": 42,
  "projection_schema_version": 2,
  "is_deleted": false
}

请求版本严格大于当前版本
→ 接受写入

请求版本小于或等于当前版本
→ 返回版本冲突

ES 当前版本 42
旧事件尝试写入 41
→ 409 Version Conflict
→ 文档不会倒退

{ "index": { "_index": "kb_gateway_wiki", "_id": "page_001", "version": 42, "version_type": "external" } }
{ "id": "page_001", "source_revision": 42, "is_deleted": false }
{ "index": { "_index": "kb_gateway_wiki", "_id": "page_002", "version": 18, "version_type": "external" } }
{ "id": "page_002", "source_revision": 18, "is_deleted": false }

errors
items[*].status
items[*].error

func HandleSearchEvent(
    ctx context.Context,
    event SearchOutboxEvent,
) error {
    page, err := pageRepo.GetIncludingDeleted(
        ctx,
        event.AggregateID,
    )
    if err != nil {
        return fmt.Errorf("load page: %w", err)
    }

    if page.Revision < event.SourceRevision {
        return fmt.Errorf(
            "source revision moved backwards: row=%d event=%d",
            page.Revision,
            event.SourceRevision,
        )
    }

    doc := BuildWikiProjection(page)

    err = searchRepo.IndexExternalVersion(
        ctx,
        "kb_gateway_wiki",
        page.ID,
        page.Revision,
        doc,
    )
    if err == nil {
        return nil
    }

    if errors.Is(err, ErrVersionConflict) {
        return verifyAlreadyApplied(
            ctx,
            page.ID,
            page.Revision,
        )
    }

    return err
}

只要 Delete 携带 external version，
任何时间到达的旧 UPSERT 都永远不能复活。

revision: 42 → 43
deleted_at: null → 删除时间

{
  "id": "page_001",
  "source_revision": 43,
  "projection_schema_version": 2,
  "is_deleted": true
}

{
  "term": {
    "is_deleted": false
  }
}

最大消息延迟
+
最大重试窗口
+
DLQ 人工重放窗口
+
安全缓冲

MySQL 已删除或不存在
但
ES 仍然 is_deleted=false

func NextRetry(attempt int) time.Duration {
    base := time.Second
    maxDelay := 30 * time.Minute

    shift := min(attempt, 10)
    delay := base * time.Duration(1<<shift)
    if delay > maxDelay {
        delay = maxDelay
    }

    jitterLimit := max(delay/5, time.Millisecond)
    jitter := time.Duration(
        rand.Int63n(int64(jitterLimit)),
    )

    return delay + jitter
}

event_id
aggregate_id
source_revision
operation
first_failed_at
last_failed_at
attempts
last_error_code
last_error_message
projection_schema_version

func ProjectionHash(doc WikiDocument) (string, error) {
    normalized := NormalizeProjection(doc)

    payload, err := json.Marshal(normalized)
    if err != nil {
        return "", err
    }

    sum := sha256.Sum256(payload)
    return hex.EncodeToString(sum[:]), nil
}

Repair 发现漂移
→ 插入 REBUILD Outbox
→ 走同一个 Projector
→ 走同一套 external version 写入

在线 Worker 一套投影
Repair 脚本另一套投影
迁移工具第三套投影

CREATE TABLE search_projection_state (
    aggregate_type              VARCHAR(64) NOT NULL,
    aggregate_id                VARCHAR(128) NOT NULL,
    source_revision             BIGINT UNSIGNED NOT NULL,
    indexed_revision            BIGINT UNSIGNED NULL,
    projection_schema_version   INT UNSIGNED NULL,
    projection_hash             CHAR(64) NULL,
    last_success_at             DATETIME(6) NULL,
    last_error_code             VARCHAR(64) NULL,
    last_error_at               DATETIME(6) NULL,
    PRIMARY KEY (
        aggregate_type,
        aggregate_id
    )
);

source_revision
→ 某个业务实体变化了多少次

projection_schema_version
→ 投影代码与字段契约是哪一版

physical_index_version
→ 当前物理索引是 v1、v2 还是 v3

page revision = 42
projection schema = 2
physical index = kb_gateway_wiki_v3

schema 1：只写 relations
schema 2：增加 relation_target_ids
schema 3：改变内容拼接规则

kb_gateway_wiki Alias
→ kb_gateway_wiki_v1

T0 记录 Outbox 水位 W0
T1 开始全量扫描
T2 page_001 更新到 revision 43
T3 全量扫描可能读到 42 或 43
T4 扫描结束
T5 回放 W0 之后的增量

{
  "actions": [
    {
      "remove": {
        "index": "kb_gateway_wiki_v1",
        "alias": "kb_gateway_wiki"
      }
    },
    {
      "add": {
        "index": "kb_gateway_wiki_v2",
        "alias": "kb_gateway_wiki",
        "is_write_index": true
      }
    }
  ]
}

require_alias=true

正常索引延迟目标
严重 Lag 告警阈值
强一致读回 MySQL 的场景
允许搜索延迟的场景

{
  "id": "page_001",
  "source_revision": 42,
  "projection_schema_version": 2
}

outbox_pending_count
outbox_oldest_pending_age_seconds
outbox_processing_count
outbox_retry_rate
outbox_dead_count
outbox_claim_timeout_count

es_bulk_item_success_rate
es_bulk_item_429_rate
es_mapping_error_rate
es_version_conflict_rate
es_write_latency_p50
es_write_latency_p95
es_write_latency_p99
es_tombstone_count

projection_revision_lag
projection_missing_count
projection_stale_count
projection_orphan_count
projection_hash_mismatch_count
projection_schema_stale_count

reindex_scanned
reindex_indexed
reindex_failed
reindex_incremental_lag
alias_current_physical_index
v1_v2_query_diff_rate

{
  "event_id": "evt-001",
  "aggregate_type": "wiki_page",
  "aggregate_id": "page_001",
  "event_revision": 41,
  "loaded_revision": 42,
  "projection_schema_version": 2,
  "target_alias": "kb_gateway_wiki",
  "operation": "UPSERT",
  "attempt": 3,
  "result": "indexed"
}

业务变化
→ MySQL 事务内递增 revision 并写 Outbox
→ Worker 读取当前事实
→ 唯一 Projector 生成完整文档
→ external version 写入稳定 Alias
→ 重试与 DLQ 保证失败可恢复
→ Tombstone 阻止旧事件复活
→ Repair 持续发现漂移
→ v1/v2 与 Alias 支持整库重建