inkreachhub/inkreach-prefect
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 写库并发控制(advisory锁串行写+分批提交+lock_timeout+死锁重试), refresh_pending写任务加重试

Browse Source 
死锁根因:sds-refresh-pending 与 label-ocr 整点并发写同一分区 Orders_2026_08,
互等对方未提交事务(ShareLock)。24h 内同类死锁 7 起。
- sync_orders_and_goods 所有写事务先取同一把 PG advisory 锁(事务级),单锁无环,死锁结构性消除
- 同步按批(1000单)独立提交,持锁窗口从整个 pending 集缩到单批
- 连接预设 lock_timeout=15s,DeadlockDetected/LockNotAvailable 整事务重试 3 次(退避1/3/7s)
- label-ocr 的 update_ocr_batch/update_ocr_fields 走同一把锁+重试
- sds/riin refresh_pending 的 upsert_records 加任务级重试(retries=3)
This commit is contained in:
yeuimu
2026-09-02 03:23:02 +00:00
parent d6be75d6b7
commit c545c1b87d
5 changed files with 103 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flows/riin/refresh_pending.py
+1 -1
View File
@@ -31,7 +31,7 @@ def fetch_pending(order_nos: list[str]) -> list:
client.close()
@task
@task(retries=3, retry_delay_seconds=[3, 6, 9])
def upsert_records(raw_orders: list) -> dict:
client = create_riin_client()
try:
flows/sds/refresh_pending.py
+1 -1
View File
@@ -32,7 +32,7 @@ def fetch_pending(order_nos: list[str]) -> list:
client.close()
@task
@task(retries=3, retry_delay_seconds=[3, 6, 9])
def upsert_records(raw_orders: list) -> dict:
client = create_sds_client()
try:
flows/shared/db.py
+99 -32
View File
@@ -8,9 +8,11 @@
from __future__ import annotations
import time
from datetime import datetime, timedelta
from typing import Any
from typing import Any, Callable
from psycopg2 import errors as pg_errors
from psycopg2.extras import execute_values
from flows.shared.mappers import GoodsUpsertInput, OrderUpsertInput
@@ -19,6 +21,14 @@ ORDER_CHUNK = 1000
GOODS_CHUNK = 2000
LOOKUP_BATCH = 10000
# 写 Orders/Goods 的互斥:所有写事务先取同一把 advisory 锁(事务级,commit/rollback 即释放)。
# 单锁无环,写事务之间从结构上不可能死锁。1768851058 = "inkr" 的 ASCII 整数表示。
DB_WRITE_LOCK_KEY = 1768851058
LOCK_TIMEOUT = "15s" # 等锁超时 → LockNotAvailable → 交给重试兜底
WRITE_RETRIES = 3 # 死锁/等锁超时的整事务重试次数
WRITE_RETRY_BACKOFF = [1, 3, 7] # 重试退避(秒)
WRITE_RETRYABLE_ERRORS = (pg_errors.DeadlockDetected, pg_errors.LockNotAvailable)
class SyncDB:
def __init__(self, block_name: str = "pg-inkreach-dmp") -> None:
@@ -27,7 +37,40 @@ class SyncDB:
self._engine = SqlAlchemyConnector.load(block_name).get_engine()
def _conn(self):
return self._engine.raw_connection()
"""取连接并预设 lock_timeout(session 级,池化复用的连接同样生效)。"""
conn = self._engine.raw_connection()
with conn.cursor() as cur:
cur.execute(f"SET lock_timeout = '{LOCK_TIMEOUT}'")
return conn
def _retryable_write(self, fn: Callable[[Any], Any]) -> Any:
"""整事务重试:仅对死锁/等锁超时(DeadlockDetected / LockNotAvailable)重试,其余异常原样抛出。
fn(conn) 负责占位锁与 commit;失败时连接已回滚,重试开新连接重跑整个事务。
"""
last_err = None
for attempt in range(WRITE_RETRIES):
conn = self._conn()
try:
result = fn(conn)
except WRITE_RETRYABLE_ERRORS as err:
conn.rollback()
conn.close()
last_err = err
if attempt == WRITE_RETRIES - 1:
raise
wait = WRITE_RETRY_BACKOFF[attempt]
print(f"[DB sync] 写库锁冲突({err.__class__.__name__}),{wait}s 后重试"
f"(第 {attempt + 2}/{WRITE_RETRIES} 次)")
time.sleep(wait)
continue
except Exception:
conn.rollback()
conn.close()
raise
conn.close()
return result
raise last_err # pragma: no cover(循环内必 return/raise)
# ---------- 查询辅助 ----------
@@ -203,11 +246,13 @@ class SyncDB:
"""更新单个订单的 ocr_* 字段。fields 为 {ocr_xxx: value} 字典。自动盖处理时间戳。"""
if not fields:
return
fields = {"ocrProcessedAt": datetime.now(), **fields}
assignments = ", ".join(f'"{k}"=%s' for k in fields)
params = list(fields.values()) + [order_no, created_at]
with self._conn() as conn:
def _once(conn) -> None:
fields_with_ts = {"ocrProcessedAt": datetime.now(), **fields}
assignments = ", ".join(f'"{k}"=%s' for k in fields_with_ts)
params = list(fields_with_ts.values()) + [order_no, created_at]
with conn.cursor() as cur:
cur.execute("SELECT pg_advisory_xact_lock(%s)", (DB_WRITE_LOCK_KEY,))
cur.execute(
f'''UPDATE "Orders" SET {assignments}
WHERE "orderNo"=%s AND "createdAt"=%s''',
@@ -215,34 +260,36 @@ class SyncDB:
)
conn.commit()
self._retryable_write(_once)
def update_ocr_batch(self, updates: list[tuple[str, datetime, dict[str, str | None]]]) -> int:
"""批量更新 ocr 字段(单连接单事务)。updates: [(orderNo, createdAt, {field: value})]。
"""批量更新 ocr 字段(单连接单事务,与其他写事务互斥)。updates: [(orderNo, createdAt, {field: value})]。
返回更新行数。
"""
if not updates:
return 0
rows = 0
with self._conn() as conn:
try:
with conn.cursor() as cur:
for order_no, created_at, fields in updates:
if not fields:
continue
fields = {"ocrProcessedAt": datetime.now(), **fields}
assignments = ", ".join(f'"{k}"=%s' for k in fields)
params = list(fields.values()) + [order_no, created_at]
cur.execute(
f'''UPDATE "Orders" SET {assignments}
WHERE "orderNo"=%s AND "createdAt"=%s''',
params,
)
rows += cur.rowcount
conn.commit()
except Exception:
conn.rollback()
raise
return rows
def _once(conn) -> int:
rows = 0
with conn.cursor() as cur:
cur.execute("SELECT pg_advisory_xact_lock(%s)", (DB_WRITE_LOCK_KEY,))
for order_no, created_at, fields in updates:
if not fields:
continue
fields = {"ocrProcessedAt": datetime.now(), **fields}
assignments = ", ".join(f'"{k}"=%s' for k in fields)
params = list(fields.values()) + [order_no, created_at]
cur.execute(
f'''UPDATE "Orders" SET {assignments}
WHERE "orderNo"=%s AND "createdAt"=%s''',
params,
)
rows += cur.rowcount
conn.commit()
return rows
return self._retryable_write(_once)
def count_ocr_status(self) -> dict[str, int]:
"""统计 OCR 处理状态(done/error/unsupported/pending)。"""
@@ -290,9 +337,30 @@ class SyncDB:
if not orders and not goods and not delete_order_nos:
print("[DB sync] 无数据可写入,跳过")
return result
return self._retryable_write(
lambda conn: self._sync_orders_and_goods_once(conn, orders, goods, delete_order_nos)
)
conn = self._conn()
def _sync_orders_and_goods_once(
self,
conn,
orders: list[OrderUpsertInput],
goods: list[GoodsUpsertInput],
delete_order_nos: list[str] | None,
) -> dict[str, int]:
"""单次写库事务:先取共享 advisory 锁与所有写事务串行化,再逐批 upsert。
每批(ORDER_CHUNK 个订单)独立提交:持锁窗口从"整个 pending 集"缩到"单批",
并发写者排队时间以秒计;事务级 advisory 锁随每批 commit 释放,单锁无环,无死锁可能。
"""
result = {
"ordersCreated": 0, "ordersUpdated": 0, "goodsCreated": 0,
"goodsUpdated": 0, "goodsRemoved": 0, "ordersDeleted": 0,
}
try:
with conn.cursor() as cur:
cur.execute("SELECT pg_advisory_xact_lock(%s)", (DB_WRITE_LOCK_KEY,))
system_name = orders[0].systemName if orders else None
all_order_nos = list({o.orderNo for o in orders} | {g.orderNo for g in goods})
@@ -344,7 +412,7 @@ class SyncDB:
print(f"[DB sync] 准备写入: {len(orders_dedup)} 订单 / {len(goods_dedup)} 商品"
+ (f" (系统={system_name})" if system_name else ""))
# 3. 分批 upsert orders + 关联 goods
# 3. 分批 upsert orders + 关联 goods(每批提交一次)
total_order_batches = -(-len(orders_dedup) // ORDER_CHUNK)
for i in range(0, len(orders_dedup), ORDER_CHUNK):
chunk = orders_dedup[i : i + ORDER_CHUNK]
@@ -356,6 +424,7 @@ class SyncDB:
stats = self._upsert_order_batch(conn, chunk, related, system_name)
for k, v in stats.items():
result[k] += v
conn.commit()
# 4. 剩余 goods(其 order 不在本次 orders 列表)
orders_set = {o.orderNo for o in orders_dedup}
@@ -381,8 +450,6 @@ class SyncDB:
except Exception:
conn.rollback()
raise
finally:
conn.close()
print(f"[DB sync] 写入完成: {result}")
return result