From 5aaddbbf7c7b634e3601ffd7782ee9c9985ca63f Mon Sep 17 00:00:00 2001 From: yeuimu <2197651308@qq.com> Date: Fri, 18 Sep 2026 10:09:24 +0000 Subject: [PATCH 1/3] =?UTF-8?q?fix:=20worker=20=E5=8A=A0=20init=20?= =?UTF-8?q?=E5=9B=9E=E6=94=B6=E5=83=B5=E5=B0=B8=E8=BF=9B=E7=A8=8B;label-oc?= =?UTF-8?q?r=20=E6=89=B9=E9=87=8F=E8=AF=B7=E6=B1=82=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E4=B8=94=E5=A4=B1=E8=B4=A5=E4=B8=8D=E8=90=BD=20ERROR?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - docker-compose: prefect-worker 加 init: true。pull 步骤的 git 子进程被杀后 无人 wait 变僵尸,9/17 宿主机重启后爆发 3.7 万僵尸占满容器 cgroup PID 配额, 导致所有 flow run "Failed to start flow run process"(全量 CRASH 约 40h)。 - label_ocr: OCR 编排服务(lm-server-canary)偶发秒级重启,原逻辑对批量请求 失败零重试并整批标永久 ERROR,9/1、9/5、9/9 三波共 ~2500 单误标。 现改为请求级重试 3 次,仍失败则保持未处理,留给下个小时 run 自动重试。 --- docker-compose.yml | 3 +++ flows/label_ocr.py | 33 +++++++++++++++++++++++++-------- 2 files changed, 28 insertions(+), 8 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 81db5f6..303a4b5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -15,6 +15,9 @@ services: prefect-worker: build: . container_name: prefect-worker + # init 进程(tini)回收孤儿/僵尸:pull 步骤的 git 子进程被杀后没人 wait 会变僵尸, + # 堆满 cgroup PID 配额后容器内无法再 fork(2026-09-17 事故根因) + init: true environment: PREFECT_API_URL: http://prefect-server:4200/api TZ: Asia/Shanghai diff --git a/flows/label_ocr.py b/flows/label_ocr.py index ff8d92a..9362d77 100644 --- a/flows/label_ocr.py +++ b/flows/label_ocr.py @@ -7,6 +7,7 @@ from __future__ import annotations import os +import time import httpx from prefect import flow, task @@ -16,6 +17,8 @@ from api.ocr import OcrClient, country_to_ocr_code from flows.shared import SyncDB, build_stats_rows, emit_sync_summary BATCH_SIZE = 200 +BATCH_RETRIES = 3 # 批量请求级重试(OCR 服务偶发秒级重启) +BATCH_RETRY_BACKOFF = [5, 15] # 重试间隔(秒) _OCR_FIELD_MAP = [ ("layout", "ocrLayout"), ("courier", "ocrCourier"), ("tracking", "ocrTracking"), @@ -63,7 +66,7 @@ def _has_running_peer() -> bool: def process_all(pending: list) -> dict[str, int]: db = SyncDB() ocr = OcrClient() - stats = {"success": 0, "error": 0, "unsupported": 0, "batches": 0} + stats = {"success": 0, "error": 0, "unsupported": 0, "batches": 0, "deferred": 0} try: # 1. 按国家分组;不支持的国家直接标 UNSUPPORTED by_country: dict[str, list[tuple[str, object, str]]] = {} @@ -89,12 +92,24 @@ def process_all(pending: list) -> dict[str, int]: urls = [{"url": url, "filename": no} for no, _, url in batch] batch_no = i // BATCH_SIZE + 1 total_batches = -(-total // BATCH_SIZE) - try: - results = ocr.recognize_batch(urls, country_code) - except Exception as err: # noqa: BLE001 - print(f"[OCR] {country_code} batch {batch_no}/{total_batches} 请求失败: {str(err)[:100]}") - db.update_ocr_batch([(no, created_map[no], {"ocrLayout": "ERROR"}) for no in created_map]) - stats["error"] += len(batch) + results = None + for attempt in range(BATCH_RETRIES): + try: + results = ocr.recognize_batch(urls, country_code) + break + except Exception as err: # noqa: BLE001 + if attempt < BATCH_RETRIES - 1: + wait = BATCH_RETRY_BACKOFF[min(attempt, len(BATCH_RETRY_BACKOFF) - 1)] + print(f"[OCR] {country_code} batch {batch_no}/{total_batches} 请求失败" + f"({str(err)[:80]}),{wait}s 后重试({attempt + 2}/{BATCH_RETRIES})") + time.sleep(wait) + else: + # OCR 服务端短暂不可用(容器秒级重启类故障):订单保持未处理, + # 留给下个小时的 run 自动重试,不落 ERROR(那会是永久终态)。 + print(f"[OCR] {country_code} batch {batch_no}/{total_batches} " + f"重试 {BATCH_RETRIES} 次仍失败,留待下次 run: {str(err)[:100]}") + if results is None: + stats["deferred"] += len(batch) continue updates = [] @@ -152,11 +167,13 @@ def label_ocr( after = db.count_ocr_status() total_processed = stats["success"] + stats["error"] + stats["unsupported"] print(f"完成{scope}: 处理 {total_processed} 单 / {stats['batches']} 批 | " - f"成功 {stats['success']} 失败 {stats['error']} 不支持 {stats['unsupported']}") + f"成功 {stats['success']} 失败 {stats['error']} 不支持 {stats['unsupported']} " + f"遗留待重试 {stats['deferred']}") print(f"累计: {after}") emit_sync_summary(f"面单 OCR{scope}", build_stats_rows("ALL", {}, [ ("本批处理", total_processed), ("成功", stats["success"]), ("失败", stats["error"]), ("不支持", stats["unsupported"]), + ("遗留待重试", stats["deferred"]), ("累计已完成", after.get("done", 0)), ("累计待处理", after.get("pending", 0)), ])) return {"processed": total_processed, **stats, **after} -- 2.54.0 From 1393b2a86bfbd6109a12d94e4bbae404317730b5 Mon Sep 17 00:00:00 2001 From: yeuimu <2197651308@qq.com> Date: Sat, 19 Sep 2026 02:57:01 +0000 Subject: [PATCH 2/3] =?UTF-8?q?fix:=20platform=5Forder=20=E6=9C=AC?= =?UTF-8?q?=E5=9C=B0=E7=9B=B4=E8=B7=91=E6=94=B9=E4=B8=BA=E7=9B=B4=E6=8E=A5?= =?UTF-8?q?=E6=89=A7=E8=A1=8C,=E4=B8=8D=E5=86=8D=20serve=20=E5=90=8C?= =?UTF-8?q?=E5=90=8D=20deployment(=E4=BC=9A=E8=A6=86=E7=9B=96=E6=AD=A3?= =?UTF-8?q?=E5=BC=8F=E9=83=A8=E7=BD=B2=E5=AF=BC=E8=87=B4=20Not=20Ready)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- flows/platform_order/riin.py | 3 ++- flows/platform_order/rouhui.py | 3 ++- flows/platform_order/sdspod.py | 3 ++- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/flows/platform_order/riin.py b/flows/platform_order/riin.py index 023797c..621e569 100644 --- a/flows/platform_order/riin.py +++ b/flows/platform_order/riin.py @@ -208,4 +208,5 @@ async def riin_platform_sync(): log_msg("===== 完成 =====") if __name__ == "__main__": - riin_platform_sync.serve(name="platform-order-riin", cron="15 * * * *") + # 直跑一次用于本地调试;勿改回 serve():同名 upsert 会覆盖 server 上挂 work pool 的正式 deployment + riin_platform_sync() diff --git a/flows/platform_order/rouhui.py b/flows/platform_order/rouhui.py index 07644c5..dc3b344 100644 --- a/flows/platform_order/rouhui.py +++ b/flows/platform_order/rouhui.py @@ -273,4 +273,5 @@ def rouhui_platform_sync(): log_msg("===== 提取完成 =====") if __name__ == "__main__": - rouhui_platform_sync.serve(name="platform-order-rouhui", cron="*/15 * * * *") \ No newline at end of file + # 直跑一次用于本地调试;勿改回 serve():同名 upsert 会覆盖 server 上挂 work pool 的正式 deployment + rouhui_platform_sync() \ No newline at end of file diff --git a/flows/platform_order/sdspod.py b/flows/platform_order/sdspod.py index 1244088..a46bc90 100644 --- a/flows/platform_order/sdspod.py +++ b/flows/platform_order/sdspod.py @@ -471,4 +471,5 @@ async def sdspod_platform_sync(): if __name__ == "__main__": - sdspod_platform_sync.serve(name="platform-order-sdspod", cron="*/15 * * * *") \ No newline at end of file + # 直跑一次用于本地调试;勿改回 serve():同名 upsert 会覆盖 server 上挂 work pool 的正式 deployment + sdspod_platform_sync() \ No newline at end of file -- 2.54.0 From f37ef24cf04c66d65ab80a47443ed48bab3374e5 Mon Sep 17 00:00:00 2001 From: Administrator <1397750220@qq.com> Date: Sun, 20 Sep 2026 16:53:49 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=95=86=E6=88=B7?= =?UTF-8?q?=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- flows/platform_order/riin.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flows/platform_order/riin.py b/flows/platform_order/riin.py index 621e569..a599430 100644 --- a/flows/platform_order/riin.py +++ b/flows/platform_order/riin.py @@ -22,7 +22,7 @@ RIIN_LOGIN_URL = "http://tshirt.riin.com/auth/api/auth/login" RIIN_ORDER_URL = "https://tshirt.riin.com/distribution/order/page" USERNAME_RIIN = "预上网" PASSWORD_RIIN = "Aa123456." -SUB_ID_LIST = ["10368", "10283", "16905", "11502", "13615", "11531"] +SUB_ID_LIST = ["10368", "10283", "16905", "11502", "13615", "11531","12279"] MAX_RETRY = 3; REQUEST_TIMEOUT = 30; PAGE_SIZE_RIIN = 200 LOOKBACK_DAYS = 6 # 回查14天避免漏单 -- 2.54.0