From eb97ee1647e6dacac3a634be5803c68d26c9725d Mon Sep 17 00:00:00 2001
From: yeuimu <2197651308@qq.com>
Date: Fri, 23 Oct 2026 10:57:40 +0000
Subject: [PATCH 1/2] =?UTF-8?q?fix:=20RIIN=20order/goods=20=E5=85=A8?=
 =?UTF-8?q?=E5=B1=80=E9=99=90=E9=80=9F(=E5=B9=B6=E5=8F=9120=E2=86=925+150m?=
 =?UTF-8?q?s=E9=97=B4=E9=9A=94,=E5=B3=B0=E5=80=BC=E2=89=A46.7QPS);?=
 =?UTF-8?q?=E6=9A=82=E5=81=9C=20recent-sync/verify=20=E8=B0=83=E5=BA=A6(?=
 =?UTF-8?q?=E5=88=86=E9=94=80=E4=BE=A7=E5=8F=8D=E9=A6=88=E8=B0=83=E7=94=A8?=
 =?UTF-8?q?=E9=87=8F=E8=BF=87=E5=A4=A7)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 flows/shared/clients.py | 20 +++++++++++++++++++-
 prefect.yaml            |  2 ++
 2 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/flows/shared/clients.py b/flows/shared/clients.py
index 74f9dc3..b414e5e 100644
--- a/flows/shared/clients.py
+++ b/flows/shared/clients.py
@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import json
+import threading
 import time
 from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, TypeVar
@@ -21,7 +22,8 @@ from flows.shared.mappers import map_sds_channel_type
 
 T = TypeVar("T")
 
-GOODS_CONCURRENCY = 20
+GOODS_CONCURRENCY = 5
+GOODS_MIN_INTERVAL_MS = 150  # 相邻两次 order/goods 调用全局最小间隔(峰值 ≤ ~6.7 次/秒)
 PAGE_DELAY_MS = 500
 MAX_RETRIES = 3
 RETRY_DELAY_MS = 3000
@@ -144,8 +146,24 @@ def fetch_sds_all_pages(client: SdsMngClient, params: dict[str, Any]) -> list[An
 
 # ---------- RIIN 商品 ----------
 
+# 全局限速:RIIN 分销侧反馈 distribution/order/goods 逐单调用对其服务器压力过大,
+# 所有线程共享节拍,相邻两次调用至少间隔 GOODS_MIN_INTERVAL_MS。
+_goods_pace_lock = threading.Lock()
+_goods_last_call_at = 0.0
+
+
+def _pace_goods_call() -> None:
+    global _goods_last_call_at
+    with _goods_pace_lock:
+        wait = _goods_last_call_at + GOODS_MIN_INTERVAL_MS / 1000 - time.monotonic()
+        if wait > 0:
+            time.sleep(wait)
+        _goods_last_call_at = time.monotonic()
+
+
 def fetch_goods_for_order(client: RiinDisClient, order_id: str) -> list[Any]:
     try:
+        _pace_goods_call()
         resp = client.get_order_goods(order_id)
         return list(resp.data)
     except Exception as err:  # noqa: BLE001
diff --git a/prefect.yaml b/prefect.yaml
index 1d1c04a..b1f2353 100644
--- a/prefect.yaml
+++ b/prefect.yaml
@@ -21,6 +21,7 @@ deployments:
   schedules:
   - cron: 30 * * * *
     timezone: Asia/Shanghai
+    active: false  # 暂停:每小时重拉近7天全部订单商品,对 RIIN distribution/order/goods 调用量过大(增量/刷新未完结/日报核验已覆盖)
   work_pool:
     name: inkreach-pool
 - name: riin-refresh-pending
@@ -39,6 +40,7 @@ deployments:
   schedules:
   - cron: 0 6 * * *
     timezone: Asia/Shanghai
+    active: false  # 暂停:与 daily-report 18:30 的核验子流程重复,保留日报一处即可
   work_pool:
     name: inkreach-pool
 - name: sds-order-sync
-- 
2.54.0


From f87662ba4b392f1bf1511406f7818be03f28f253 Mon Sep 17 00:00:00 2001
From: yeuimu <2197651308@qq.com>
Date: Fri, 23 Oct 2026 11:06:49 +0000
Subject: [PATCH 2/2] =?UTF-8?q?fix:=20order/goods=20=E9=99=90=E9=80=9F?=
 =?UTF-8?q?=E6=94=B6=E7=B4=A7=E8=87=B3=E5=B3=B0=E5=80=BC<3=E6=AC=A1/?=
 =?UTF-8?q?=E7=A7=92(340ms+0~160ms=E9=9A=8F=E6=9C=BA=E9=97=B4=E9=9A=94)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 flows/shared/clients.py | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/flows/shared/clients.py b/flows/shared/clients.py
index b414e5e..ab8d673 100644
--- a/flows/shared/clients.py
+++ b/flows/shared/clients.py
@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import json
+import random
 import threading
 import time
 from datetime import datetime, timedelta, timezone
@@ -23,7 +24,8 @@ from flows.shared.mappers import map_sds_channel_type
 T = TypeVar("T")
 
 GOODS_CONCURRENCY = 5
-GOODS_MIN_INTERVAL_MS = 150  # 相邻两次 order/goods 调用全局最小间隔(峰值 ≤ ~6.7 次/秒)
+GOODS_MIN_INTERVAL_MS = 340  # 相邻两次 order/goods 调用全局最小间隔(峰值 < 3 次/秒)
+GOODS_MAX_JITTER_MS = 160  # 在最小间隔上叠加的随机抖动上限,实际间隔 340~500ms 随机
 PAGE_DELAY_MS = 500
 MAX_RETRIES = 3
 RETRY_DELAY_MS = 3000
@@ -147,7 +149,7 @@ def fetch_sds_all_pages(client: SdsMngClient, params: dict[str, Any]) -> list[An
 # ---------- RIIN 商品 ----------
 
 # 全局限速:RIIN 分销侧反馈 distribution/order/goods 逐单调用对其服务器压力过大,
-# 所有线程共享节拍,相邻两次调用至少间隔 GOODS_MIN_INTERVAL_MS。
+# 所有线程共享节拍,相邻两次调用间隔 = GOODS_MIN_INTERVAL_MS + 随机抖动(340~500ms,峰值 < 3 次/秒)。
 _goods_pace_lock = threading.Lock()
 _goods_last_call_at = 0.0
 
@@ -155,7 +157,8 @@ _goods_last_call_at = 0.0
 def _pace_goods_call() -> None:
     global _goods_last_call_at
     with _goods_pace_lock:
-        wait = _goods_last_call_at + GOODS_MIN_INTERVAL_MS / 1000 - time.monotonic()
+        interval = (GOODS_MIN_INTERVAL_MS + random.uniform(0, GOODS_MAX_JITTER_MS)) / 1000
+        wait = _goods_last_call_at + interval - time.monotonic()
         if wait > 0:
             time.sleep(wait)
         _goods_last_call_at = time.monotonic()
-- 
2.54.0