yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(rag_client): params读写列不一致致重复行——_load_cached_key按id查但写入id=getID()随机值,缓存永远miss→每次发key新增一行(实测user-01堆12行);ON DUPLICATE KEY UPDATE因params_name无唯一索引永不触发。修:读改按params_name+读-改-写幂等(存在则更新首行删多余行)

Browse Source
This commit is contained in:
ymq 2026-09-12 08:04:17 +08:00
parent 78c89a0cce
commit c32ceaa920
1 changed files with 20 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
pipeline_service/rag_client.py
View File

@ -119,7 +119,7 @@ async def _ensure_downapp(sor):
async def _load_cached_key(sor, user_id): async def _load_cached_key(sor, user_id):
pname = _KEY_PARAM_PREFIX + str(user_id) pname = _KEY_PARAM_PREFIX + str(user_id)
recs = await sor.sqlExe( recs = await sor.sqlExe(
"SELECT params_value FROM params WHERE id=${n}$", {"n": pname}) "SELECT params_value FROM params WHERE params_name=${n}$ LIMIT 1", {"n": pname})
await sor.sqlExe("COMMIT", {}) await sor.sqlExe("COMMIT", {})
if not recs: if not recs:
return "" return ""
@ -160,10 +160,25 @@ async def get_owner_apikey(owner_id):
# RC4 加密存 params（与平台密码字段约定一致） # RC4 加密存 params（与平台密码字段约定一致）
pname = _KEY_PARAM_PREFIX + str(owner_id) pname = _KEY_PARAM_PREFIX + str(owner_id)
enc = _key_encode(key) enc = _key_encode(key)
await sor.sqlExe( # 幂等按 params_name 读-改-写：params 表 params_name 无唯一索引，
"INSERT INTO params (id, params_name, params_value) VALUES (${i}$, ${n}$, ${v}$) " # 不能用 INSERT...ON DUPLICATE KEY UPDATE（永不触发→重复插行）。
"ON DUPLICATE KEY UPDATE params_value=${v}$", # 旧 bug：_load_cached_key 按 id 查、这里按 getID() 随机 id 写，缓存永远 miss
{"i": getID(), "n": pname, "v": enc}) # →每次发 key 都新增一行（实测 user-01 堆了 12 行）。2026-09-12 修。
exists = await sor.sqlExe(
"SELECT id FROM params WHERE params_name=${n}$ ORDER BY id LIMIT 1", {"n": pname})
await sor.sqlExe("COMMIT", {})
if exists:
# 收敛历史重复行：更新首行，删多余行
await sor.sqlExe(
"UPDATE params SET params_value=${v}$ WHERE id=${i}$",
{"v": enc, "i": exists[0].id})
await sor.sqlExe(
"DELETE FROM params WHERE params_name=${n}$ AND id<>${i}$",
{"n": pname, "i": exists[0].id})
else:
await sor.sqlExe(
"INSERT INTO params (id, params_name, params_value) VALUES (${i}$, ${n}$, ${v}$)",
{"i": getID(), "n": pname, "v": enc})
await sor.sqlExe("COMMIT", {}) await sor.sqlExe("COMMIT", {})
_key_cache[owner_id] = key _key_cache[owner_id] = key
return key, "" return key, ""