diff --git a/AGENTS.md b/AGENTS.md index 41d1670..4290d7a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -71,7 +71,7 @@ uv run prefect # 运行 prefect CLI - 同步任务(对齐 inkreach-bi-auth-mng 的 sync tasks)放在 `flows/riin/` 和 `flows/sds/`,每系统各 6 个: - `order_sync`(按时间窗口)、`incremental_sync`(按 DB 最近下单时间增量,无参)、 `recent_sync`(近 N 天)、`refresh_pending`(刷新未完结)、`by_orders`(RIIN 指定单号)/ `refresh_100_prefix`(SDS 100 前缀)、`verify`(逐天核验修复)。 -- 共享层在 `flows/shared/`:`mappers.py`(RIIN/SDS→统一 Order/Goods)、`db.py`(PG upsert `SyncDB`)、`clients.py`(客户端工厂+顺序分页+重试+渠道字典)。 +- 共享层在 `flows/shared/`:`mappers.py`(RIIN/SDS→统一 Order/Goods)、`db.py`(PG upsert `SyncDB`,写并发控制:advisory 锁串行 + 分批提交 + lock_timeout + 死锁重试)、`clients.py`(客户端工厂+顺序分页+重试+渠道字典)。 - 本地直接跑(调试):`uv run python flows/riin/.py` - 部署到 work pool:`uv run prefect deploy --all`(按 `prefect.yaml`,全部 12 个 deployment 挂 `inkreach-pool`)。 - 手动触发一次:`uv run prefect deployment run '/'` diff --git a/docs/dev/systematic-design.md b/docs/dev/systematic-design.md index 5e5736a..20703f8 100644 --- a/docs/dev/systematic-design.md +++ b/docs/dev/systematic-design.md @@ -124,6 +124,7 @@ inkreach-prefect/ - `mappers.py`:RIIN/SDS 原始订单 → 统一 `OrderUpsertInput` / `GoodsUpsertInput`,含状态/平台/快递来源/国家归一映射。 - `db.py`:`SyncDB.sync_orders_and_goods()` —— Orders/Goods 批量 `ON CONFLICT` upsert,**写入前用库中已有 `createdAt` 覆盖**(保证幂等),Goods 写前删除该订单不再出现的行;另有 `get_latest_ordered_at` / `get_pending_orders` / `get_order_count_by_status` / `get_local_order_nos` / `get_local_goods_count` / `get_orders_with_100_prefix` 等查询。 + - **写并发控制**(2026-08 死锁修复):所有写 Orders/Goods 的事务(同步 flow 与 label-ocr 的 OCR 回写)先取同一把 PG advisory 锁(`DB_WRITE_LOCK_KEY`,事务级,commit/rollback 即释放)串行化——单锁无环,写事务之间结构性无死锁;`sync_orders_and_goods` 按批(`ORDER_CHUNK`=1000)独立提交,持锁窗口从"整个 pending 集"缩到"单批";连接预设 `lock_timeout=15s`;对 `DeadlockDetected` / `LockNotAvailable` 整事务自动重试 3 次(退避 1/3/7s),另有 flow/task 级 retries 兜底。 - `clients.py`:客户端工厂(读 Secret Block)、`fetch_riin/sds_all_pages`(顺序分页,页间 500ms,仅对 5xx/429/超时重试)、RIIN 商品并发拉取(并发 20)、SDS 物流渠道字典(24h 缓存)。 ### 关键差异点 diff --git a/flows/riin/refresh_pending.py b/flows/riin/refresh_pending.py index 851d886..7f76e86 100644 --- a/flows/riin/refresh_pending.py +++ b/flows/riin/refresh_pending.py @@ -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: diff --git a/flows/sds/refresh_pending.py b/flows/sds/refresh_pending.py index 1a3345e..d2f5e01 100644 --- a/flows/sds/refresh_pending.py +++ b/flows/sds/refresh_pending.py @@ -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: diff --git a/flows/shared/db.py b/flows/shared/db.py index 02c1e00..b76a998 100644 --- a/flows/shared/db.py +++ b/flows/shared/db.py @@ -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