Python后端中間件專(zhuān)題20工單已創(chuàng)建卻搜不到——refresh、異步索引與版本亂序用戶(hù)剛提交“VPN 憑據(jù)失效”詳情頁(yè)馬上能打開(kāi)搜索頁(yè)卻暫時(shí)找不到?？头詾楣蝸G了后端看見(jiàn) PostgreSQL 行、Outbox 事件、Worker 日志都在懷疑是搜索延遲。這時(shí)最糟糕的“修復(fù)”是讓創(chuàng)建接口等待 ES 刷新并在 ES 不可用時(shí)把數(shù)據(jù)庫(kù)事務(wù)也判失敗這會(huì)把一個(gè)可重建的投影升格成業(yè)務(wù)事實(shí)。這一課用一條時(shí)間線(xiàn)把四個(gè)時(shí)刻分開(kāi)數(shù)據(jù)庫(kù)提交、事件發(fā)布、投影寫(xiě)入、搜索 refresh 可見(jiàn)再讓兩個(gè) Worker 亂序完成驗(yàn)證舊版本不能回蓋新版本。上一課練習(xí)答案答案 EX-19-01 [code]保存為cursor_guard_probe.py從project/運(yùn)行fromticketflow.search.queryimport(InvalidCursor,SearchFilters,build_search_body,decode_cursor,encode_cursor,)sort_values[2.5,2026-09-21T09:00:0000:00,ticket-9]cursorencode_cursor(sort_values)assertdecode_cursor(cursor)sort_values bodybuild_search_body(tenant_idtenant-a,queryVPN,filtersSearchFilters(),size10,cursorcursor,)assertbody[search_after]sort_valuesprint(cursor round-trip ok)try:decode_cursor(not-base64)exceptInvalidCursor:print(invalid cursor rejected)else:raiseAssertionError(invalid cursor was accepted)從project/運(yùn)行命令python cursor_guard_probe.py預(yù)期輸出cursor round-trip ok invalid cursor rejected注意這個(gè)測(cè)試證明當(dāng)前非法樣本被拒并不證明所有畸形/超長(zhǎng)游標(biāo)都經(jīng)充分校驗(yàn)Base64 更不是授權(quán)邊界。任何后續(xù)頁(yè)面請(qǐng)求仍由服務(wù)端加租戶(hù)過(guò)濾。答案 EX-19-02 [prose]v1 的游標(biāo)保存的是 v1 命中列表最后一條的評(píng)分、時(shí)間、ID切到 v2 后Mapping、分析器、數(shù)據(jù)回填時(shí)間甚至評(píng)分都可能變了。游標(biāo)可解碼只說(shuō)明它是三元組不說(shuō)明它在 v2 的相同位置。簡(jiǎn)單產(chǎn)品策略是在索引 generation 變化時(shí)返回“結(jié)果已更新請(qǐng)重新搜索”不偷偷接著翻頁(yè)。更強(qiáng)的策略是用 PIT 鎖定同一 concrete index 的讀取視圖在游標(biāo)中綁定 PIT ID/查詢(xún)指紋設(shè)置過(guò)期和清理此方案成本更高必須處理 PIT 失效。當(dāng)前課程模塊沒(méi)有 PIT不應(yīng)聲稱(chēng)跨切換無(wú)重?zé)o漏。四個(gè)時(shí)刻不要用“創(chuàng)建成功”合并t0 POST /tickets → PostgreSQL 中 ticket Outbox 同事務(wù)提交 → HTTP 201 t1 relay 發(fā)布事件到 RabbitMQ 并拿到 confirm t2 Worker 讀取最新 ticket向 ES 寫(xiě)入投影 t3 ES refresh 后這份 document 對(duì) _search 可見(jiàn)t0是事實(shí)提交詳情頁(yè)應(yīng)以它為準(zhǔn)t3是讀投影可見(jiàn)可能晚于t0。在t2后馬上用_get和_search得到不同觀(guān)察并不矛盾近實(shí)時(shí)搜索依賴(lài) refresh。默認(rèn)每次寫(xiě)都強(qiáng)制refreshtrue會(huì)讓批量索引付出顯著代價(jià)TicketFlow 的project_ticket()明確傳refreshFalse讓 ES 以正常刷新節(jié)奏處理。本地用一個(gè)最小實(shí)驗(yàn)核對(duì)“寫(xiě)請(qǐng)求不強(qiáng)制刷新”和“版本隨請(qǐng)求傳遞”但它不模擬真實(shí) ES refreshimportasynciofromdatetimeimportdatetime,timezonefromticketflow.search.projectorimportTicketProjection,project_ticketclassRecordingIndexer:def__init__(self):self.calls[]asyncdefindex(self,**kwargs):self.calls.append(kwargs)asyncdefmain():backendRecordingIndexer()ticketTicketProjection(ticket_idticket-20,tenant_idtenant-a,titleVPN 憑據(jù)失效,body登錄報(bào)錯(cuò),statusopen,priorityhigh,tags(vpn,),version3,updated_atdatetime(2026,9,22,tzinfotimezone.utc),)awaitproject_ticket(backend,aliasticketflow-tickets,ticketticket)assertbackend.calls[0][version]3assertbackend.calls[0][version_type]external_gteassertbackend.calls[0][refresh]isFalseprint(version3 external_gte refreshFalse)asyncio.run(main())保存為projection_probe.py從project/運(yùn)行命令python projection_probe.py預(yù)期輸出version3 external_gte refreshFalse真實(shí)服務(wù)上應(yīng)記錄 Outbox/Worker 的處理時(shí)間并有界輪詢(xún)_search看到 t2 之后短暫未命中是預(yù)期窗口超過(guò)預(yù)定延遲閾值才報(bào)警。不要把“目前無(wú)結(jié)果”直接等同于“索引服務(wù)不可用”二者在 HTTP 契約上不同。兩個(gè) Worker 反著到誰(shuí)說(shuō)了算假設(shè)工單從 version 1 改到 2、再改到 3。消息至少一次投遞兩個(gè) Worker 可以分別拿著 version 3 與 version 2 的投影完成順序是 3→2。如果只按“最后到達(dá)”覆蓋搜索會(huì)倒退。project_ticket()將 PostgreSQL 的版本傳給 ESPUT /ticketflow-tickets/_doc/ticket-20?version3version_typeexternal_gte PUT /ticketflow-tickets/_doc/ticket-20?version2version_typeexternal_gte第二次寫(xiě)收到 version conflict409適配器把它當(dāng)作舊事件的預(yù)期終態(tài)而非反復(fù)重試的 Worker 故障。相等版本允許重復(fù)寫(xiě)入是external_gte的語(yǔ)義因此同一個(gè) version 必須對(duì)應(yīng)確定的業(yè)務(wù)快照如果生產(chǎn)系統(tǒng)允許同版本不同內(nèi)容那應(yīng)重新設(shè)計(jì)版本/冪等合同。這里有一處值得留意的細(xì)節(jié)Worker 實(shí)際如何從事件取版本、是否讀取最新 PostgreSQL 快照、索引 doc 與 Outbox 關(guān)聯(lián)都要用真實(shí)端到端旅程觀(guān)察。本課 unit 測(cè)試只證明project_ticket()發(fā)出的參數(shù)與 HTTP adapter 對(duì) 409 的處理不代替 Worker、broker、ES 的集成證據(jù)。跑本課檢查點(diǎn)并知道它的證明范圍從課程根運(yùn)行python tools/verify_executable_revisions.py--lesson 20--python python正式本地命令對(duì) lesson-20 獨(dú)立安裝并執(zhí)行 baseline 與 09–20 全部歷史合同包括投影 document、外部版本與refreshFalse。遠(yuǎn)端 selectortests/integration/test_search_elasticsearch.py::test_real_refresh_window_and_external_version_rejects_stale_event在唯一 ES8 索引記錄寫(xiě)后、顯式 refresh 前的可見(jiàn)性不強(qiáng)求每次都不可見(jiàn)核對(duì)寫(xiě)請(qǐng)求未強(qiáng)制 refresh、刷新后可搜索以及遲到舊版本不能覆蓋新版本。它不模擬兩個(gè)真實(shí) Worker 亂序未運(yùn)行仍是 PENDING。若生產(chǎn)中“長(zhǎng)時(shí)間搜不到”沿著ticket → outbox pending → relay confirm → worker effect → ES alias/doc → refresh/_search逐站定位。若版本反退先查是否遺漏 external version、是否向錯(cuò)誤 index alias 寫(xiě)入、是否多個(gè)系統(tǒng)獨(dú)立產(chǎn)生 version。若索引 HTTP 500不能按 409 舊事件吞掉它應(yīng)進(jìn)入有限重試/DLQ 與告警。本課練習(xí)練習(xí) EX-20-01 [code]寫(xiě)一段httpx.MockTransport實(shí)驗(yàn)讓 ES 對(duì) version 3 返回 201、對(duì)遲到 version 2 返回 409。用ElasticsearchHTTP.index()兩次提交同一 ID斷言請(qǐng)求都帶version_typeexternal_gte最終打印new accepted; stale ignored。第 21 課給出完整代碼和命令。練習(xí) EX-20-02 [prose]列出創(chuàng)建工單后 t0–t3 四個(gè)時(shí)刻如果 t1 后 Worker 暫停而搜索沒(méi)有命中詳情頁(yè)與搜索頁(yè)各應(yīng)怎樣表現(xiàn)解釋為什么不能讓“搜索尚未可見(jiàn)”回滾已經(jīng)提交的工單。下一站新 Mapping 如何上線(xiàn)第 21 課不刪業(yè)務(wù)數(shù)據(jù)、不直接修改正在讀的索引從 PostgreSQL 回填新 concrete index核對(duì)文檔集合再原子切換 alias。這個(gè)過(guò)程也會(huì)暴露“搜索是可重建投影”真正的操作成本。本課完整核心模塊search/projector.py以下是lesson-20的完整投影寫(xiě)入模塊。完整代碼說(shuō)明了輸入形狀、版本仲裁參數(shù)和刷新策略閱讀時(shí)請(qǐng)對(duì)照上面的四時(shí)刻不要把a(bǔ)wait index()誤讀成搜索已經(jīng)立即可見(jiàn)。from__future__importannotationsfromdataclassesimportdataclassfromdatetimeimportdatetimefromtypingimportProtocolclassAsyncIndexer(Protocol):asyncdefindex(self,**kwargs:object)-None:...dataclass(frozenTrue,slotsTrue)classTicketProjection:ticket_id:strtenant_id:strtitle:strbody:strstatus:strpriority:strtags:tuple[str,...]version:intupdated_at:datetimeasyncdefproject_ticket(indexer:AsyncIndexer,*,alias:str,ticket:TicketProjection)-None:Write a search projection without allowing an older event to win. Outbox delivery is at-least-once and different workers can finish out of order. Elasticsearch external versioning turns the PostgreSQL ticket version into the conflict arbiter; the queue arrival time is irrelevant. awaitindexer.index(indexalias,idticket.ticket_id,document{ticket_id:ticket.ticket_id,tenant_id:ticket.tenant_id,title:ticket.title,body:ticket.body,status:ticket.status,priority:ticket.priority,tags:list(ticket.tags),version:ticket.version,updated_at:ticket.updated_at.isoformat(),},versionticket.version,version_typeexternal_gte,refreshFalse,)