🛑 擋下用 Upsert

修 Race Condition 的 PR

🔧 正確的修法

🎯 什麼情境該想到我

當你的流程是「寫 DB + 發一則 MQ / event」,而兩者可能不同步時(訊息比 commit 早到、或 commit 成功但發送失敗)。

症狀:worker 收到 message,去 DB 卻查不到對應資料。

⚙️ 怎麼用(三個層次,由淺到深)

  1. 最小修法:把發送移出 transaction,commit 成功後才發。
    const order = await prisma.$transaction(tx => tx.order.create({ data }))
    await sendMessageToMq(order.id) // commit 之後才發送
    → 仍留邊界:commit 成功但 send 失敗(crash/網路),訊息遺失。
  2. Outbox Pattern(嚴謹解):把「業務寫入」與「事件記錄」放同一個 transaction:
    await prisma.$transaction(async tx => {
      const order = await tx.order.create({ data })
      await tx.outbox.create({ data: { topic: "order.created", payload: { orderId: order.id } } })
    })
    // 另有獨立 relay(polling 或 CDC)讀 outbox 表再發到 MQ
    → DB commit 與 event 產生變成原子操作,要嘛都有要嘛都沒有。

★ 為什麼「commit 後再發」仍然不夠

BEGIN; INSERT order; COMMIT;     ✅ 訂單存了
★ 這裡程序被 kill / 機器斷電 / 網路斷
publish(event)                   ❌ 永遠不會執行
→ ★ 訂單存在,但下游永遠不知道

★ 這不是「機率很低」的問題,是「一定會發生」的問題。
每天十萬筆訂單,就算只有萬分之一,每天也有 10 筆靜默遺失——
而且你不會知道,直到客戶來抱怨。

反過來「先發再寫」也不行:訊息發了但交易失敗 → 下游收到不存在的訂單的事件。

★ 這就是雙寫問題(dual write):兩個獨立的系統無法原子地一起成功或一起失敗。

★ Outbox 的核心:把「跨系統」化約成「單一資料庫的本地交易」

BEGIN;
  INSERT INTO orders (...) VALUES (...);
  INSERT INTO outbox (id, aggregate_id, type, payload, created_at)
       VALUES (gen_random_uuid(), :order_id, 'OrderPlaced', :json, now());
COMMIT;                          -- ★ 要嘛都成功,要嘛都失敗

★ 而單一資料庫的交易是可靠的(見 工具-交易與隔離等級)——
問題被化約成了一個已經被解決的問題。

★ Relay 的兩種實作

① 輪詢 outbox 表② ★ CDC(讀 WAL)
做法SELECT * FROM outbox WHERE published_at IS NULL LIMIT 100Debezium 讀 PG 邏輯解碼 / MySQL binlog
延遲輪詢間隔(通常 100ms–1s)★ 接近即時
對主庫負載★ 有(持續查詢 + 更新)★ 幾乎零(讀 WAL 不碰資料表)
運維★ 簡單(就是一段程式)要跑 Debezium/Kafka Connect
順序保證要自己處理★ 天然有序(WAL 就是順序)
建議★ 中小規模的預設規模大或已有 Kafka 生態時

★ 輪詢版的關鍵細節:

-- ★ 用 FOR UPDATE SKIP LOCKED 讓多個 relay 實例可以並行而不重複
SELECT * FROM outbox
 WHERE published_at IS NULL
 ORDER BY created_at
 LIMIT 100
 FOR UPDATE SKIP LOCKED;

SKIP LOCKED 是這裡的關鍵——它讓多個 worker 各自取走不同的一批,不會互相阻塞也不會重複。

★ 必須配套的三件事

事項為什麼
★ 消費端冪等Outbox 是至少一次:relay 發送成功但標記失敗 → 會重發(見 工具-MQ消費端防禦三原則)
★ 清理舊資料outbox 表會無限成長 → 定期刪除已發布且超過保留期的(★ 或用分區表按時間 DROP)
★ 監控積壓未發布訊息的最舊時間 是關鍵指標。relay 掛了要立刻知道

★ 常見的實作錯誤

錯誤後果
★ relay 用「先發送再標記」發送成功但標記失敗 → 重發(可接受,靠冪等)。
★ 反過來「先標記再發送」會丟訊息——這是嚴重錯誤
★ payload 只存 ID 不存內容消費端拿到 ID 回頭查 → 查到的是「現在」的狀態,不是「事件發生當下」的狀態
沒有 aggregate_id無法做分區與順序保證
outbox 和業務表在不同資料庫★ 完全失去意義——那又變回雙寫了
沒有清理表膨脹 → 查詢變慢 → relay 越來越慢

★ 什麼時候不需要 Outbox

情況替代
★ 訊息遺失可接受(遙測、分析、通知)直接發,失敗就算了
下游可以主動來查★ 輪詢 / API 查詢——不需要推送就不需要保證投遞
只有一個下游且能容忍延遲定期批次同步
★ 根本不該跨服務合併回同一個交易(見 工具-服務邊界與演進式拆分)

🧪 我實際套用的紀錄

  • 2026-07-13:(待填)

⚠️ 注意 / 什麼時候不適用

  • ★★ outbox 表必須和業務表在「同一個資料庫」,否則完全沒有意義。
  • ★ relay 一定要「先發送、後標記」(見上)。順序反了會丟訊息。
  • ★ 消費端冪等不是可選的。
  • outbox 寫入會增加主交易的成本(多一次 INSERT)。高頻寫入時要量,必要時用分區表。
  • ★ 一定要監控積壓。relay 悄悄掛掉 → 訊息全部堆在表裡沒人發現,幾小時後才發現下游什麼都沒收到。
  • 事件內容要是「當時的快照」,不要只存 ID(見上)。
  • CDC 的 replication slot 會洩漏:CDC 工具離線 → ★ 主庫的 WAL 無法回收 → 磁碟爆掉
    (見 工具-預寫日誌與崩潰復原)。要監控 slot 落後量。
  • ★ 這解決的是「投遞」不是「處理」。下游收到了但處理失敗,是另一個問題(DLQ,見 工具-訊息傳遞整合模式)。

🔗 相關工具

🎯 什麼情境該想到我

分工:這張是消費端的即用清單。各模式的完整展開(含 Kafka、RabbitMQ、SQS 的行為差異)→ 冪等接收者、死信通道、保證投遞;用資料庫條件更新做冪等的完整做法 → 工具-用條件更新做狀態機冪等。

當你在寫 message queue 的 consumer / worker,要讓它在真實世界(重複投遞、replica 延遲)下不出錯時。

⚙️ 怎麼用(三件事一起做)

  1. Idempotent(冪等):MQ 是 at-least-once,重複投遞是常態 → 同一則訊息處理多次,結果要一致。
  2. Retry with backoff / requeue:查不到資料時退避重試或重新入列,因為可能是 replication lag 或極端時序 → 絕不自己造資料補上。
  3. 搭配上游修好時序:下游防禦不能取代上游正確性,兩邊都要做(上游見 工具-Transactional-Outbox-Pattern)。
層級做法解決
上游Outbox pattern時序錯誤、事件遺失
下游Retry with backoffreplication lag、極端時序
消費端Idempotent 處理at-least-once 重複投遞

★ 冪等的兩種做法(Go 骨架)

做法 A:記錄已處理的訊息 ID(與業務寫入在同一個交易裡)

func (h *Handler) Handle(ctx context.Context, msg Message) error {
	tx, err := h.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
 
	// 同一則訊息第二次進來時,這裡影響 0 列 → 直接視為成功
	res, err := tx.ExecContext(ctx,
		`INSERT INTO processed_messages (id) VALUES ($1) ON CONFLICT DO NOTHING`, msg.ID)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return nil // 已處理過
	}
 
	if err := h.apply(ctx, tx, msg); err != nil { // 業務邏輯,用同一個 tx
		return err
	}
	return tx.Commit()
}

關鍵是同一個交易:如果「記錄 ID」和「業務寫入」分開提交,兩者之間當機,就會出現「記了 ID 但沒做事」或「做了事但沒記 ID」。

做法 B:讓操作本身冪等——用條件更新推進狀態機,例如 UPDATE orders SET status='shipped' WHERE id=$1 AND status='paid',第二次執行影響 0 列。詳見 工具-用條件更新做狀態機冪等。

★ 重試要有上限與退避

要素為什麼
指數退避(1s、2s、4s…)讓下游有時間恢復,也讓複本延遲有時間追上
隨機抖動避免大量消費者在同一時刻一起重試,造成新的尖峰
最大次數確定性的錯誤重試再多次也不會成功
超過上限 → 死信佇列不阻塞後面的訊息,也不讓失敗的訊息消失;有人工檢查的入口

判斷該不該重試:暫時性錯誤(網路逾時、複本上查不到剛寫入的資料、下游 503)重試;確定性錯誤(格式錯誤、業務規則不允許)直接送死信,重試只是浪費時間。

★ 「查不到資料」的三種可能

消費者收到訊息,但資料庫裡查不到對應的資料:

  1. 複本延遲:讀的是複本,主庫已經寫入但還沒同步 → 退避重試,或改讀主庫。
  2. 生產端時序錯誤:訊息在交易提交之前就送出了 → 根本的修法在生產端(Outbox)。
  3. 交易被回滾:訊息送出了,但交易最後失敗 → 這則訊息本來就不該存在,同樣要靠 Outbox 修。

三種情況都不該在消費端自己補一筆資料——那會製造出殘缺的「幽靈資料」(見 工具-區分修復與掩蓋症狀)。

🧪 我實際套用的紀錄

  • 2026-07-13:(待填)

⚠️ 注意 / 什麼時候不適用

  • 「worker 找不到資料 → 自己補一筆」是反模式,會造出殘缺/幽靈資料。

🔗 相關工具

🔍 怎麼審這種修法

🎯 什麼情境該想到我

分工:這張是判準卡。本例中 MQ 消費端的正確做法 → 工具-MQ消費端防禦三原則;從生產端根治 → 工具-Transactional-Outbox-Pattern。

當你在 review 一個「看起來合理」的修法(upsert、retry、try-catch、加鎖…),想判斷它是真的修好還是只是把症狀藏起來時。

⚙️ 怎麼用

  1. 先問「為什麼這裡會有這個問題?」 而不是「這個修法對不對」。
    • 例:不是問「upsert 寫得對嗎」,而是「為什麼這裡會需要 upsert」。
  2. 順著症狀跨系統往上追因果:worker 症狀 → MQ 時序 → sender 的 transaction 邊界。
  3. 判斷準則:如果這個修法讓 error 消失、但根本的錯誤狀態仍然存在(只是不再噴錯),它就是掩蓋,不是修復。
  4. 掩蓋型修法要擋下來——症狀消失後就沒人會再發現真正的問題。

★ 常見的「看起來像修復」的做法,怎麼判斷

做法什麼時候是掩蓋什麼時候是修復
upsert(查不到就新增)資料「應該已經存在」卻查不到——你在用新資料蓋掉一個時序錯誤業務上本來就允許「第一次寫入」與「後續更新」走同一條路
retry錯誤是確定性的(邏輯錯、資料錯),重試只是延後失敗錯誤是暫時性的(網路抖動、複本延遲),而且有上限與退避
try/catch 吞掉錯誤吞掉之後回傳一個「看起來正常」的值在系統邊界把錯誤轉成明確的回應,並記錄下來
加鎖用一把大鎖蓋住一個不該並發的設計保護一個確實需要互斥的共享狀態,且範圍最小
加大 timeout真正的問題是慢查詢或死結下游的正常延遲分布本來就比原設定長
重啟服務記憶體或連線洩漏,重啟只是讓它晚點再爆一次性的損壞狀態,且已經找到並修掉原因

★ 四個檢查問題

  1. 錯誤狀態還在嗎? 修法讓錯誤訊息消失了,但系統裡的資料或狀態仍然是錯的——那就是掩蓋。
  2. 能解釋為什麼會發生嗎? 講不出因果鏈(「A 在 B 之前發生,因為……」),就還沒找到根因。
  3. 下一個人還看得到問題嗎? 修法讓問題在監控、日誌、測試裡都消失了——之後就沒有人會再發現它。
  4. 換一個輸入還會發生嗎? 寫一個能重現原問題的測試;修復之後它應該通過,而且是因為對的理由通過。

★ 緊急止血時的正確做法

事故當下,掩蓋型的做法可能是對的——先讓使用者不再受影響。但要同時:

  • 在程式碼裡標清楚:// TODO(#1234): 暫時的 upsert,根因是 sender 在 commit 前發送訊息。
  • 開一張追蹤根因的工單,並在事後檢討中列為行動項(見 工具-無指責事後檢討)。
  • 讓暫時的做法可觀測:每次走到那條補丁路徑都記一筆指標,才知道根因修好後它是否真的不再被觸發。

🧪 我實際套用的紀錄

  • 2026-07-13:(待填)

⚠️ 注意 / 什麼時候不適用

  • 緊急止血時可先用掩蓋型修法擋住,但必須開票追 root cause,別讓它變成永久補丁。
  • 核心心法:不要比 AI 更會 coding,要比它更會 reasoning。

🔗 相關工具