Python后端中间件专题21新 Mapping 上线不删库——alias 重建与切换“给标题换一个分析器”看似是一行 Mapping 配置实际上旧索引已经按旧规则倒排好了词项。原地修改并不会让历史文档自动重新分词。更危险的是直接删旧索引再建新索引会让搜索在回填期间完全空白还会失去比较和回滚的锚点。TicketFlow 的做法是把ticketflow-tickets当作稳定的读取 alias先在另一边建ticketflow-tickets-v2在一个明确的维护窗口里锁住工单写入从 PostgreSQL 回填、刷新、核对 ID 与版本再用一次_aliases操作切过去。业务真相仍在 PostgreSQL这不是迁移工单表。上一课练习答案答案 EX-20-01 [code]保存为version_http_probe.py。我们用httpx.MockTransport替代外部网络只测 TicketFlow 发出的真实 HTTP 请求与 409 处理不伪称 ES 的版本引擎已经运行。importasyncioimporthttpxfromticketflow.search.http_adapterimportElasticsearchHTTPasyncdefmain():seen[]defrespond(request:httpx.Request)-httpx.Response:seen.append(request)returnhttpx.Response(201ifrequest.url.params[version]3else409)asyncwithhttpx.AsyncClient(base_urlhttp://es.test:9200,transporthttpx.MockTransport(respond),)asclient:backendElasticsearchHTTP(client,aliasticketflow-tickets)forversionin(3,2):awaitbackend.index(indexticketflow-tickets,idticket-20,document{tenant_id:tenant-a,title:VPN},versionversion,version_typeexternal_gte,)assert[request.url.params[version]forrequestinseen][3,2]assertall(request.url.params[version_type]external_gteforrequestinseen)print(new accepted; stale ignored)asyncio.run(main())从project/运行命令python version_http_probe.py预期输出new accepted; stale ignored这个适配器把 HTTP 409 当“旧事件已经没有必要再写”的成功终态而不是一般的 5xx。生产上还要确认 409 的确来自 version conflict不能把其他 409 一概吞掉。后续验收应以真实 ES 的旧版本写入和最终 document 为准。答案 EX-20-02 [prose]t0PostgreSQL 同事务提交 ticket 与 OutboxAPI 返回 201t1relay 发布并收到 broker confirmt2Worker 将投影文档写给 ESt3ES refresh 后_search可见。若 t1 后 Worker 暂停详情从 PostgreSQL 继续返回 200 和最新版本搜索暂时未命中是投影落后应有延迟指标/告警。不能回滚 t0它已经提交ES 又不是同一个 ACID 事务的参与者把独立依赖的短故障说成“工单创建失败”只会制造重复提交风险。一次安全重建的操作顺序假设当前 alias 指向ticketflow-tickets-v1目标是-v2。操作员先确认源数据范围、磁盘空间、旧索引保留窗口和回滚条件然后PostgreSQL 事务执行LOCK TABLE tickets IN SHARE MODE。普通读继续INSERT/UPDATE/DELETE等持有ROW EXCLUSIVE的写入等待create_index(v2, ticket_index_definition())不触碰 v1在同一持锁事务内读取工单事实按 ID 写入 v2 并显式_refresh比较候选索引的 ID 与version。缺失、意外 ID 或同 ID 旧版本都阻止切换再读一次完整源投影即使调用者错误地声称已暂停写入只要两次快照不同也拒绝切换在_aliases的一个请求里移除 v1 alias 并添加 v2随后提交数据库事务、释放写锁返回之前不删除 v1用真实搜索样本、租户负例与错误率观察新版本需要回滚就把 alias 指回保留的 v1。这是课程项目刻意选择的“受控暂停写入”方案实现简单、边界可验证代价是大索引回填期间写接口会等待所以不适合无停机的大规模生产。那类系统应改用 Outbox 水位追平或 v1/v2 双写并在切换点验证追平。本课不能一边允许活跃写入一边只核对数量和 ID 后宣称零丢更新。从程序的拒绝路径读设计意图rebuild_and_switch()首先要求调用者明确证明写入已冻结然后检测源 ID 是否重复。候选索引回填后它比较actual_versions与事实库的 ID/version再二次读取源投影。如果 v2 少了ticket-2抛出ReindexMismatch(missing1, unexpected0)如果 ID 相同但 version 落后则抛stale1两种情况都不会调用switch_alias()。下面是可运行的安全闸门实验保存为reindex_guard_probe.pyimportasynciofromticketflow.search.reindeximportReindexMismatch,rebuild_and_switchclassIncompleteBackend:def__init__(self):self.switchedFalseasyncdefcreate_index(self,name,definition):passasyncdefbulk_index(self,name,documents):passasyncdefdocument_versions(self,name):return{ticket-1:1}# 故意漏掉 ticket-2asyncdefswitch_alias(self,alias,new_index):self.switchedTruereturnticketflow-tickets-v1asyncdefmain():backendIncompleteBackend()try:documents[{id:ticket-1,version:1},{id:ticket-2,version:1},]awaitrebuild_and_switch(backend,aliasticketflow-tickets,new_indexticketflow-tickets-v2,source_readerlambda:documents,definition{},source_writes_frozenTrue,)exceptReindexMismatchaserror:print(str(error))else:raiseAssertionError(incomplete index unexpectedly switched)assertbackend.switchedisFalseprint(alias still points to v1)asyncio.run(main())从project/运行命令python reindex_guard_probe.py预期输出candidate projection mismatch: missing1, unexpected0 alias still points to v1IncompleteBackend是故意不完整的边界替身只验证“缺文档时不可切换”的程序决策真实 ES adapter、PostgreSQL 读取和_aliases原子操作需要远端检查。运行检查点与部署前预检先从课程根运行本课 revision verifierpython tools/verify_executable_revisions.py--lesson 21--python python再切换到project/单独运行重建脚本的 dry-run$env:PYTHONPATH (Resolve-Path.\src).Path python scripts/rebuild_search.py--new-indexticketflow-tickets-v2--dry-run本地 unit 覆盖成功顺序、缺文档、旧版本、回填期源变化与“未声明写冻结”--dry-run只检验命名/模式。远端 selectortests/integration/test_search_elasticsearch.py::test_real_postgres_lock_rebuild_alias_switch_and_rollback仅在授权隔离项目中生成唯一 alias/v1/v2 和测试工单用两条 PostgreSQL 连接及有限 lock/statement timeout 证明竞争写被挡住再经真实 ES 回填、ID/version 校验、切换及回滚最后仅清理这次索引和行。它没有操作共享ticketflow-ticketsalias。未执行远端时锁、真实重建和回滚均是 PENDING不能凭本地 fake 宣称通过。失败定位顺序创建失败看 index 名和 Mappingbulk 有 item errors 看单条字段类型不要只看 HTTP 200ID 集合不齐查数据库读取窗口与 refresh/scrollalias 指向多个索引则先停止切换并厘清所有者切换后效果差先把 alias 指回保留的 v1再比较查询样本与版本不要删 v2 以消灭证据。本课练习练习 EX-21-01 [code]写reindex_duplicate_probe.py用 fake backend 调rebuild_and_switch()让source_reader返回两行相同id且带version并传source_writes_frozenTrue断言抛ReindexMismatch(source contains duplicate document ids)且create_index()和switch_alias()都没调用。打印duplicate source blocked before create。第 22 课给出完整代码。练习 EX-21-02 [prose]解释新索引回填完成、alias 切换之前有一张工单更新时为什么“数量和 ID 全相同”仍可能搜索到旧字段给出至少一种追平方案并说明回滚 alias 为什么不会回滚 PostgreSQL 的工单修改。下一课如果整个 ES 都不可用alias 重建可控制计划内变更但不能保证依赖永远在线。第 22 课会把 ES 503 变成明确的搜索故障同时让详情和写入继续依赖 PostgreSQL 事实。本课完整核心模块search/reindex.py以下为lesson-21完整模块。与运行脚本分工此模块实现校验和切换顺序scripts/rebuild_search.py才负责真实 HTTP adapter、读取事实库和 CLI 安全参数。两者配合使用不能只复制本模块就宣称有可运行重建命令。from__future__importannotationsfromdataclassesimportdataclassfromcollections.abcimportCallablefromtypingimportProtocolclassReindexMismatch(RuntimeError):A candidate projection differs from the PostgreSQL source set.classReindexBackend(Protocol):asyncdefcreate_index(self,name:str,definition:dict[str,object])-None:...asyncdefbulk_index(self,name:str,documents:list[dict[str,object]])-None:...asyncdefdocument_versions(self,name:str)-dict[str,int]:...asyncdefswitch_alias(self,alias:str,new_index:str)-str|None:...dataclass(frozenTrue,slotsTrue)classReindexResult:previous_index:str|Nonenew_index:strdocument_count:intasyncdefrebuild_and_switch(backend:ReindexBackend,*,alias:str,new_index:str,source_reader:Callable[[],list[dict[str,object]]],definition:dict[str,object],source_writes_frozen:boolFalse,)-ReindexResult:Build and validate a concrete index before changing the read alias. Alias switching is safe only while ticket writes are actually blocked. The production CLI holds a PostgreSQL table lock for this entire coroutine; the explicit flag prevents a caller from accidentally omitting that boundary. We read the source again before switching and compare complete projection documents, then compare candidate versions rather than just document IDs. ifnotsource_writes_frozen:raiseReindexMismatch(ticket writes must be frozen during backfill and alias switch)documentssource_reader()expected_by_id:dict[str,dict[str,object]]{}fordocumentindocuments:identitystr(document[id])ifidentityinexpected_by_id:raiseReindexMismatch(source contains duplicate document ids)expected_by_id[identity]dict(document)awaitbackend.create_index(new_index,definition)awaitbackend.bulk_index(new_index,documents)actual_versionsawaitbackend.document_versions(new_index)actual_idsset(actual_versions)expected_idsset(expected_by_id)ifactual_ids!expected_ids:missinglen(expected_ids-actual_ids)unexpectedlen(actual_ids-expected_ids)raiseReindexMismatch(fcandidate projection mismatch: missing{missing}, unexpected{unexpected})stalesum(1foridentity,documentinexpected_by_id.items()ifactual_versions.get(identity)!int(document[version]))ifstale:raiseReindexMismatch(fcandidate projection version mismatch: stale{stale})final_documentssource_reader()final_by_id{str(document[id]):dict(document)fordocumentinfinal_documents}iflen(final_by_id)!len(final_documents):raiseReindexMismatch(source contains duplicate document ids)iffinal_by_id!expected_by_id:raiseReindexMismatch(source changed during backfill; alias was not switched)previousawaitbackend.switch_alias(alias,new_index)returnReindexResult(previous,new_index,len(documents))