Compare commits

...

3 Commits

Author SHA1 Message Date
root 8feb38db37 fix: 资源包查询修复翻页死循环并加 Status 过滤
- debug_volc_sync.py 缺少 seen_tokens 防循环保护,火山 API 翻页到头后
  NextToken 循环回起始 token,导致无限翻页(实测 900s 拉 1893 页假象,
  真实数据仅 8 页 137 个包)
- 两个 fetch_all_packages 请求加 Status=Effective,服务端直接过滤
  UsedUp/Expired/Refunded 等余额为 0 的无用包(实测过滤后 0 余额包为 0)
- 主脚本循环检测日志从 DEBUG 提升为 WARNING,循环收尾不再静默
2026-08-11 03:39:57 +00:00
root e29d94f9e8 feat: 日志统一加 [VolcSync] 前缀便于筛选
模块级重绑定 print(一处覆盖全部 52 处调用, 含 log()/dprint/裸print),
traceback.print_exc 改走带前缀通道, 去除 log() 内嵌标识避免双前缀。

筛选: docker logs litellm 2>&1 | grep VolcSync
2026-08-11 02:53:33 +00:00
root 729e5c7bc3 fix: 限额峰值改为首锁+余额门控重锁,修复越算越多
原 Smart Peak 逻辑把每次用量增长误判为余额恢复(Jump)锁更高峰值,
叠加火山 AvailableAmount 更新滞后,stable_limit 当天一路抬升
(399,836 -> 599,511)且闸门永不触发,额度实际被烧穿。

新策略(volc:peak_pool 当日键):
- 首次同步锁定当日峰值(usage + balance)
- 仅 balance 自身净上涨(新资源包/对账回补 >=50k)按发放增量抬峰
- pool 回落时 Drop 收紧,永不回升
- usage 驱动的 pool 上涨一律不得抬高限额
新增 volc:last_balance 键记录上次余额快照
2026-08-11 02:53:18 +00:00
2 changed files with 89 additions and 27 deletions
+15 -1
View File
@@ -116,12 +116,15 @@ def fetch_all_packages():
version = "2022-01-01" version = "2022-01-01"
all_packages = [] all_packages = []
current_token = "" current_token = ""
seen_tokens = set()
while True: while True:
try: try:
payload = { payload = {
"ResourceType": "Package", "ResourceType": "Package",
"MaxResults": "20", "MaxResults": "20",
# 只拉生效中的数据包,跳过 UsedUp/Expired/Refunded 等余额为 0 的无用包
"Status": "Effective",
"NextToken": current_token, "NextToken": current_token,
} }
headers, body = get_signed_headers(action, version, payload) headers, body = get_signed_headers(action, version, payload)
@@ -134,8 +137,19 @@ def fetch_all_packages():
if items: if items:
all_packages.extend(items) all_packages.extend(items)
next_token = data.get("Result", {}).get("NextToken", "") next_token = data.get("Result", {}).get("NextToken", "")
if not next_token or next_token == current_token:
# Infinite Loop Protection: VolcEngine API 翻页到头后 NextToken 会
# 循环回起始 token(实测:数据只有 ~8 页,第 9 页绕回第一个 token),
# 没有该保护会无限翻页。
if not next_token:
break break
if next_token == current_token:
break
if next_token in seen_tokens:
print(f"WARNING: Detected pagination cycle with token {next_token}. Stopping.")
break
seen_tokens.add(next_token)
current_token = next_token current_token = next_token
time.sleep(0.2) time.sleep(0.2)
except Exception as e: except Exception as e:
+73 -25
View File
@@ -41,6 +41,27 @@ except ImportError:
pass pass
# ==========================================
# 🏷️ Unified Log Tag (统一日志标识,便于过滤)
# ==========================================
# 本脚本所有输出统一带 [VolcSync] 前缀,方便日志筛选:
# docker logs litellm 2>&1 | grep VolcSync
# 实现方式:模块级重绑定 print(`print` 在模块命名空间中按调用时查找),
# 一处覆盖 log()/dprint 与全部裸 print 调用;对模块外部无任何影响。
LOG_TAG = "[VolcSync]"
_original_print = print
def _tagged_print(*args, **kwargs):
if args and isinstance(args[0], str):
args = (args[0].lstrip("\n"), *args[1:])
_original_print(LOG_TAG, *args, **kwargs)
print = _tagged_print
# ========================================== # ==========================================
# 🛠️ Configuration & Constants # 🛠️ Configuration & Constants
# ========================================== # ==========================================
@@ -157,7 +178,7 @@ def log(msg, level="INFO"):
icon = {"DEBUG": "🔍", "INFO": "ℹ️", "WARNING": "⚠️", "ERROR": "❌"}.get( icon = {"DEBUG": "🔍", "INFO": "ℹ️", "WARNING": "⚠️", "ERROR": "❌"}.get(
level, "📝" level, "📝"
) )
print(f"{icon} [VolcSync] {msg}", flush=True) print(f"{icon} {msg}", flush=True)
def dprint(msg): def dprint(msg):
@@ -842,6 +863,10 @@ def fetch_all_packages():
payload = { payload = {
"ResourceType": "Package", "ResourceType": "Package",
"MaxResults": "20", "MaxResults": "20",
# 服务端只返回生效中的数据包,跳过 UsedUp/Expired/Refunded 等
# 余额为 0 的无用包,显著减少翻页次数与 QPS 占用。
# 周期型包(月度重置)状态仍为 Effective,不会被误过滤。
"Status": "Effective",
"NextToken": current_token, "NextToken": current_token,
} }
headers, body = get_signed_headers(action, version, payload) headers, body = get_signed_headers(action, version, payload)
@@ -855,14 +880,17 @@ def fetch_all_packages():
next_token = data.get("Result", {}).get("NextToken", "") next_token = data.get("Result", {}).get("NextToken", "")
# Infinite Loop Protection # Infinite Loop Protection
# 火山 API 翻页到头后 NextToken 会循环回起始 token(实测数据只有
# ~8 页时第 9 页绕回第一个 token),必须用 seen_tokens 拦下载断。
if not next_token: if not next_token:
break break
if next_token == current_token: if next_token == current_token:
break break
if next_token in seen_tokens: if next_token in seen_tokens:
if DEBUG_MODE: log(
print( f"⚠️ VolcEngine pagination cycle detected (token {next_token} repeated). "
f"DEBUG: Detected pagination cycle with token {next_token}. Stopping." f"Stopping after {len(all_packages)} packages.",
level="WARNING",
) )
break break
@@ -1308,21 +1336,32 @@ def active_monitor_loop():
dynamic_key = f"limit:dynamic:volcengine:remaining:{target}" dynamic_key = f"limit:dynamic:volcengine:remaining:{target}"
# ========================================== # ==========================================
# 🏔️ Smart Peak Tracking (Auto-Correction) # 🏔️ Daily Peak Lock (First-Sync + Balance-Gated Re-Lock)
# ========================================== # ==========================================
# Strategy: # 背景:AvailableAmount 是各资源包实例的实时剩余,但火山统计有
# 1. Capture Jumps (Recovery): If Balance jumps UP, trust it immediately. (Fixes latency) # 更新滞后(高用量时段会短暂冻结,实测 20 万 token 消耗内余额不动)。
# 2. Capture Drops (Expiry): If Balance drops HUGE (>20%), trust it. (Fixes expiry) # 旧逻辑把每次用量增长误判为"余额恢复"(Jump)锁更高峰值,
# - Why trust drops? Because if it's a false drop (glitch), Step 1 will fix it # 导致 stable_limit 越算越多(当天 399,836 -> 599,511 一路抬升)。
# automatically on the next sync when it "Jumps" back up. # 修正策略(日内限额单调可控):
# 1. 首次同步锁定当日峰值(= usage + balance 的总配额估计);
# 2. 仅当 balance 自身净上涨(新资源包发放/对账回补,≥50k)
# 时,按新增额度抬高峰值 —— 幅度 = 历史峰值 + 发放增量,
# 绝不把滞后虚高的 usage 一起算进去;
# 3. balance 回落(额度扣减追上用量)时 Drop 收紧;
# 4. usage 增长驱动的 pool 上涨一律不得抬高限额。
# 实测:协作奖励计划包每自然日发放、偶发一天 2 个,故必须有第 2 条;
# 次日 00:00 (CST) 自动重建当日峰值。
total_pool = current_usage + balance total_pool = current_usage + balance
peak_key = f"volc:peak_pool:{today_str}:{target}" peak_key = f"volc:peak_pool:{today_str}:{target}"
# Short-term memory (Last seen value) # Short-term memory (Last seen value) - 仅用于日志对比/增量计算
last_pool_key = f"volc:last_seen_pool:{target}" last_pool_key = f"volc:last_seen_pool:{target}"
last_pool_val = redis_client.get(last_pool_key) last_pool_val = redis_client.get(last_pool_key)
last_pool = float(last_pool_val) if last_pool_val else 0.0 last_pool = float(last_pool_val) if last_pool_val else 0.0
last_balance_key = f"volc:last_balance:{target}"
last_balance_val = redis_client.get(last_balance_key)
last_balance = float(last_balance_val) if last_balance_val else 0.0
# Long-term memory (Day's High Water Mark) # Long-term memory (Day's High Water Mark)
old_peak_val = redis_client.get(peak_key) old_peak_val = redis_client.get(peak_key)
@@ -1330,21 +1369,31 @@ def active_monitor_loop():
current_peak = historical_peak current_peak = historical_peak
# 1. JUMP DETECTION (Recharge / Latency Resolved) # 1. INITIALIZATION (First sync of the day): lock once
# If pool jumps significantly (>5%) from last seen, it's a fresh, true value. if historical_peak == 0:
if total_pool > (last_pool * 1.05) and last_pool > 0: if DEBUG_MODE and last_pool > 0:
if DEBUG_MODE:
print( print(
f"DEBUG: 📈 Balance Jump ({int(last_pool):,} -> {int(total_pool):,}) - Locking New Peak.", f"DEBUG: 📌 Locking Daily Peak ({int(last_pool):,} -> {int(total_pool):,}) - First Sync Today.",
flush=True, flush=True,
) )
current_peak = total_pool current_peak = total_pool
redis_client.setex(peak_key, 172800, int(current_peak)) redis_client.setex(peak_key, 172800, int(current_peak))
# 2. DROP DETECTION (Resource Expiry) # 2. RECHARGE (Balance-driven upward re-lock)
# If pool drops (even small amounts), we accept it. # 新资源包发放 / API 对账回补 → balance 自身净上涨 ≥50k。
# Why? Because keeping a phantom high peak is dangerous. # 只把"发放增量"加进已锁定的峰值,不重算整个滞后 pool。
# If it was a glitch, Rule #1 will fix it next time it bounces back. elif last_balance > 0 and (balance - last_balance) >= 50000:
grant_delta = int(balance - last_balance)
current_peak = historical_peak + grant_delta
if DEBUG_MODE:
print(
f"DEBUG: 🆕 New Quota Detected (balance {int(last_balance):,} -> {int(balance):,}, +{grant_delta:,}) - Re-locking Daily Peak to {int(current_peak):,}.",
flush=True,
)
redis_client.setex(peak_key, 172800, int(current_peak))
# 3. DROP DETECTION (Resource Expiry / latency catch-up)
# pool 回落时收紧限额;回落之后保持不变,绝不回升。
elif total_pool < historical_peak and historical_peak > 0: elif total_pool < historical_peak and historical_peak > 0:
drop_ratio = ( drop_ratio = (
historical_peak - total_pool historical_peak - total_pool
@@ -1357,13 +1406,12 @@ def active_monitor_loop():
current_peak = total_pool current_peak = total_pool
redis_client.setex(peak_key, 172800, int(current_peak)) redis_client.setex(peak_key, 172800, int(current_peak))
# 3. INITIALIZATION / DRIFT # 4. NO USAGE-DRIVEN RE-LOCK: 用量增长(balance 不变)不得抬高限额。
elif historical_peak == 0 or total_pool > historical_peak: # (旧逻辑的 Bug 就在这里:把滞后虚高当余额恢复。)
current_peak = total_pool
redis_client.setex(peak_key, 172800, int(current_peak))
# Always update "Last Seen" for next comparison # Always update "Last Seen" for next comparison
redis_client.setex(last_pool_key, 86400, int(total_pool)) redis_client.setex(last_pool_key, 86400, int(total_pool))
redis_client.setex(last_balance_key, 86400, int(balance))
# Calculate Stable Limit # Calculate Stable Limit
safety_margin = max(current_peak * 0.15, 250000) safety_margin = max(current_peak * 0.15, 250000)
@@ -1472,7 +1520,7 @@ def active_monitor_loop():
except Exception as e: except Exception as e:
print(f"Error in monitor loop / 监控循环出错: {e}", flush=True) print(f"Error in monitor loop / 监控循环出错: {e}", flush=True)
traceback.print_exc() print(traceback.format_exc(), end="")
time.sleep(5) time.sleep(5)