fix(automation): 重登结算改用新事务读取,修掉「成功被记成失败」
线上第一轮自动重登(10 条)报告「成功 0 / 失败 10」,但库里这 10 条的 attempt_status 全是 success —— 账号其实都救回来了,是结算读错了。 根因是 MySQL 的隔离级别:默认 **REPEATABLE READ**。调度器传进来的会话在挑候选 时就已经开了事务,而结算复用了这个会话去读 runner(另一个会话)刚提交的结果; 同一事务里重新查询只会看到事务开始时的快照,于是全部读成「未成功」。SQLite 的 读一致性与 MySQL 不同,所以这类问题在 SQLite 上跑一万遍也复现不出来。 后果不只是统计错:失败的账号会被写退避时间,其中 2 条还被判成「极验」进了待 人工队列 —— 等于把救回来的账号又标成需要人工。 改动: - 新增 settle_fresh_results():自己新开一个会话读结果并结算,绝不复用调度器的 事务;run_ck_relogin 改用它。 (另一路 SQLite→MySQL 的测试改造已把测试套件真正跑在 MySQL 上,本次新增的回归 测试 test_settlement_sees_other_session_commit 正是靠这一点才守得住:它先让一个 会话开着事务,再由另一个会话提交成功,断言结算必须读到成功。)
This commit is contained in:
@@ -114,6 +114,34 @@ def build_select_query(
|
||||
).limit(max(int(job.batch_size or 0), 1))
|
||||
|
||||
|
||||
def settle_fresh_results(
|
||||
task_ids: list[int], *, now: datetime, max_attempts: int = 2
|
||||
) -> dict[str, int]:
|
||||
"""在**新会话(新事务)**里读回批次结果并结算。
|
||||
|
||||
必须新开会话,不能复用调度器传进来的 ``db``:那个会话在挑候选时就已经开了
|
||||
事务,而 MySQL 默认隔离级别是 **REPEATABLE READ** —— 在同一事务里重新查询
|
||||
只会看到事务开始时的快照,看不到 runner 在另一个会话里提交的成功结果。
|
||||
|
||||
线上为此踩过一次:10 条重登**实际全部成功**,结算却按失败写回,给它们加了
|
||||
退避时间,其中 2 条还被标成「极验」进了待人工队列。SQLite 的读一致性与 MySQL
|
||||
不同,这类问题在 SQLite 上复现不出来。
|
||||
"""
|
||||
session = SessionLocal()
|
||||
try:
|
||||
rows = session.query(LoginTask).filter(LoginTask.id.in_(task_ids)).all()
|
||||
counters = settle_results(
|
||||
{row.id: row for row in rows},
|
||||
task_ids,
|
||||
now=now,
|
||||
max_attempts=max_attempts,
|
||||
)
|
||||
session.commit()
|
||||
return counters
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
def count_pending_relogin(db: Session, *, max_attempts: int = 2) -> int:
|
||||
"""待自动重登的**积压**条数(给页面显示用)。
|
||||
|
||||
@@ -211,15 +239,8 @@ def run_ck_relogin(db: Session, job: AutomationJob) -> dict[str, Any]:
|
||||
finally:
|
||||
runner_db.close()
|
||||
|
||||
# runner 用的是另一个会话,这里必须重新查,不能读旧对象。
|
||||
db.expire_all()
|
||||
fresh = {
|
||||
row.id: row
|
||||
for row in db.query(LoginTask).filter(LoginTask.id.in_(task_ids)).all()
|
||||
}
|
||||
|
||||
counters = settle_results(fresh, task_ids, now=now, max_attempts=max_attempts)
|
||||
db.commit()
|
||||
# runner 在另一个会话里提交了结果:必须用新事务读回(见 settle_fresh_results)。
|
||||
counters = settle_fresh_results(task_ids, now=now, max_attempts=max_attempts)
|
||||
|
||||
parts = [
|
||||
f"本轮重登 {counters['scanned']} 条",
|
||||
|
||||
Reference in New Issue
Block a user