From f25c2a0aa5936e5600c2e4ab0a985a4e01336279 Mon Sep 17 00:00:00 2001 From: guke Date: Thu, 6 Aug 2026 17:49:58 +0800 Subject: [PATCH] =?UTF-8?q?fix(auto-exchange):=20=E3=80=8C=E5=BD=93?= =?UTF-8?q?=E5=A4=A9=E5=B7=B2=E5=85=91=E3=80=8D=E6=A0=87=E8=AE=B0=E6=8C=81?= =?UTF-8?q?=E4=B9=85=E5=8C=96=E5=88=B0=20app=5Fconfig=EF=BC=8C=E9=98=B2?= =?UTF-8?q?=E9=9D=9E0=E7=82=B9=E9=87=8D=E5=90=AF=E9=87=8D=E5=A4=8D?= =?UTF-8?q?=E8=A1=A5=E6=89=AB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit worker 原用内存变量 last_run 记当天是否已兑，进程重启即归零；叠加「启动首轮无条件跑」， 导致每次非 0 点部署/重启都全量补扫一遍，把 0 点后才达标的用户（0 点是零头、白天攒够； 或 0 点后才赚到金币）在非 0 点兑成现金——即用户反馈的「非 0 点也出现金币转现金记录」。 改为把「上次成功兑换的北京日」持久化到 app_config(key=auto_exchange.last_run_date，仿 compare_alert 水位范式)：当天已兑则同一北京日内重启/部署整轮跳过，不再补扫；真漏了 0 点 (标记<今天，如服务器宕机)才补跑一轮，保留原 Persistent 补跑语义。逐用户幂等保留做兜底。 新增 tests/test_daily_auto_exchange.py：首轮兑换+落库标记、同日重启不重复补扫(回归)、跨天补跑。 Co-Authored-By: Claude Opus 4.8 --- app/core/daily_exchange_worker.py | 63 ++++++++++++++--- deploy/daily-exchange.md | 6 +- tests/test_daily_auto_exchange.py | 113 ++++++++++++++++++++++++++++++ 3 files changed, 171 insertions(+), 11 deletions(-) create mode 100644 tests/test_daily_auto_exchange.py diff --git a/app/core/daily_exchange_worker.py b/app/core/daily_exchange_worker.py index 24b22e4..03431c2 100644 --- a/app/core/daily_exchange_worker.py +++ b/app/core/daily_exchange_worker.py @@ -5,9 +5,14 @@ `wallet.daily_auto_exchange`。 健壮性: -- **逐用户幂等**:当天已有 exchange_in 流水的用户跳过(见 wallet._has_exchange_in_on), - 故启动补跑 / 多次唤醒 / 进程重启都安全,不会重复兑。 -- **当天首跑即补**:进程起来时若当天还没兑过,立即兑一轮(等价原 timer 的 Persistent 补跑)。 +- **持久化「当天已兑」标记**(app_config `auto_exchange.last_run_date`):当天已兑则整轮跳过, + **标记跨进程重启不丢** → 同一北京日内多次启动/部署不再重复补扫。这是修 bug 的关键:原来用 + 内存变量记「今天跑过没」,进程重启即归零,导致每次非 0 点部署都全量补扫一遍、把 0 点后才达标 + 的用户在**非 0 点**兑成现金(用户反馈的「非 0 点也出现金币转现金记录」)。 +- **漏了 0 点即补**:标记 < 今天(如 0 点服务器宕机)时标记 != today → 照常补跑一轮(保留原 + timer 的 Persistent 语义)。这类补跑确实落非 0 点,但仅限「真漏了 0 点」的罕见场景,非常态。 +- **逐用户幂等**:当天已有 exchange_in 流水的用户跳过(见 wallet._has_exchange_in_on),做兜底 + 防同机多进程 / 标记写入失败的竞态,不会重复兑。 - **同机多进程互斥**:文件锁保证多 worker 只有一个实际跑(防跨进程并发导致 TOCTOU 双兑)。 - **开关**:settings.AUTO_EXCHANGE_ENABLED=false 时不启动(与脚本/原 timer 同一开关)。 """ @@ -27,10 +32,14 @@ from sqlalchemy.exc import SQLAlchemyError from app.core import rewards from app.core.config import settings from app.db.session import SessionLocal +from app.models.app_config import AppConfig from app.repositories import wallet as wallet_repo logger = logging.getLogger("shagua.daily_exchange") _LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "daily_exchange.lock" +# 「上次成功自动兑的北京日」标记,持久化在 app_config(仿 compare_alert 水位,不进 CONFIG_DEFS—— +# 它是 worker 内部运行状态,非运营可配项)。用它替代内存 last_run:重启不丢 → 同日不重复补扫。 +LAST_RUN_KEY = "auto_exchange.last_run_date" def _touch_lock() -> None: @@ -72,9 +81,44 @@ def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]: _LOCK_PATH.unlink() -def _exchange_once() -> dict: +def _read_last_run(db) -> date | None: + """读持久化的「上次成功自动兑的北京日」标记(app_config,跨进程重启不丢)。脏值当作没跑过。""" + row = db.get(AppConfig, LAST_RUN_KEY) + if row is None or not row.value: + return None + try: + return date.fromisoformat(row.value) + except (ValueError, TypeError): + return None + + +def _write_last_run(db, day: date) -> None: + """落库「当天已兑」标记。用专用 key 直接写 AppConfig(仿 compare_alert 水位,不走 set_value)。""" + iso = day.isoformat() + row = db.get(AppConfig, LAST_RUN_KEY) + if row is None: + db.add(AppConfig(key=LAST_RUN_KEY, value=iso, updated_by_admin_id=None)) + else: + row.value = iso + db.commit() + + +def _exchange_if_due(db, today: date) -> dict | None: + """当天未兑过(持久化标记 != today)才跑一轮并记标记;已兑过返回 None(整轮跳过)。 + + 标记写在 daily_auto_exchange 之后:即便中途崩,标记仍是旧值 → 下次重跑,逐用户幂等会跳过已兑的、 + 补完剩下的(安全)。真漏了 0 点(标记 < today)时标记 != today,仍会补跑,保留原「当天首跑即补」。 + """ + if _read_last_run(db) == today: + return None + result = wallet_repo.daily_auto_exchange(db) + _write_last_run(db, today) + return result + + +def _exchange_once(today: date) -> dict | None: with SessionLocal() as db: - return wallet_repo.daily_auto_exchange(db) + return _exchange_if_due(db, today) async def _run_loop() -> None: @@ -89,16 +133,15 @@ async def _run_loop() -> None: async def _run_locked_loop(interval: int) -> None: logger.info("daily auto-exchange worker started interval=%ss", interval) - # 本进程上次跑过的北京日;None=尚未跑过本进程(启动即补当天)。 - last_run: date | None = None try: while True: try: _touch_lock() today = rewards.cn_today() - if last_run != today: - result = await asyncio.to_thread(_exchange_once) - last_run = today + # 「今天是否已兑」以持久化标记为准(见 _exchange_if_due),不再用内存变量 → + # 进程重启不会把当天当「没跑过」重复补扫;已兑当天返回 None(整轮跳过)。 + result = await asyncio.to_thread(_exchange_once, today) + if result is not None: logger.info("daily auto-exchange done date=%s result=%s", today, result) except SQLAlchemyError: logger.exception("daily auto-exchange db error") diff --git a/deploy/daily-exchange.md b/deploy/daily-exchange.md index 794257f..f59355f 100644 --- a/deploy/daily-exchange.md +++ b/deploy/daily-exchange.md @@ -5,9 +5,13 @@ ## 现状(2026-06 起):已改为 App 进程内任务,不再需要 systemd timer 「0 点自动兑换」现由 **App 进程内后台任务** `app.core.daily_exchange_worker` 负责:App 一起来就 -每 10 分钟检查、跨过北京 0 点自动跑一轮(逐用户幂等、文件锁互斥、重启补跑当天遗漏)。**无需再装 / 启用 +每 10 分钟检查、跨过北京 0 点自动跑一轮(逐用户幂等、文件锁互斥)。**无需再装 / 启用 `daily-exchange.timer`**。 +- **「当天已兑」标记持久化在 app_config**(key=`auto_exchange.last_run_date`):当天兑过后,同一北京日 + 内进程重启 / 部署**不再重复补扫**——避免把 0 点后才达标的用户在非 0 点兑现金。只有真漏了 0 点 + (标记 < 今天,如 0 点服务器宕机)才会在重启后补跑一轮。 + - 开关仍是 `.env` 的 `AUTO_EXCHANGE_ENABLED`(false → worker 不启动);间隔由 `AUTO_EXCHANGE_CHECK_INTERVAL_SEC`(默认 600s=10min)控。 - 下方 systemd timer / 脚本属**遗留 + 手动应急**:逻辑同一套且幂等,可手动 `--once` 补跑; 但**不要再 `enable` timer 与进程内 worker 并存**(虽幂等不会双兑,纯属多余)。 diff --git a/tests/test_daily_auto_exchange.py b/tests/test_daily_auto_exchange.py new file mode 100644 index 0000000..c9ba774 --- /dev/null +++ b/tests/test_daily_auto_exchange.py @@ -0,0 +1,113 @@ +"""0 点自动兑金币 worker:持久化「当天已兑」标记,防同日重启重复补扫。 + +背景(路径 B bug):worker 原用内存变量 last_run 记「今天跑过没」,进程重启即归零 → 每次 +非 0 点部署/重启都会全量补扫一遍,把 0 点后才达标的用户在**非 0 点**兑成现金(用户反馈的 +「非 0 点也出现金币转现金记录」)。修复:标记持久化到 app_config,跨重启不丢。 +""" +from __future__ import annotations + +from datetime import date + +import pytest +from sqlalchemy import delete, select, update + +from app.core import daily_exchange_worker as w +from app.db.session import SessionLocal +from app.models.app_config import AppConfig +from app.models.user import User +from app.models.wallet import CashTransaction, CoinAccount +from app.repositories import wallet as wallet_repo + +_PHONE_SEQ = [0] + + +@pytest.fixture(autouse=True) +def _isolate_exchange_state(): + """daily_auto_exchange 全表扫描 + app_config 标记会跨用例泄漏(SQLite 测试库 session 级共享, + commit 后 rollback 是 no-op),故每个用例先清零所有余额 + 清掉自动兑标记,保证干净起步。""" + db = SessionLocal() + try: + db.execute(update(CoinAccount).values( + coin_balance=0, cash_balance_cents=0, invite_cash_balance_cents=0)) + db.execute(delete(AppConfig).where(AppConfig.key == w.LAST_RUN_KEY)) + db.commit() + finally: + db.close() + yield + + +def _new_user(db, *, coin: int) -> int: + """建一个 User + 指定金币余额的 CoinAccount,返回 user_id。""" + _PHONE_SEQ[0] += 1 + # 198 段避免撞其他测试文件的固定手机号 / username UNIQUE + u = User(phone=f"198{_PHONE_SEQ[0]:08d}", username=f"ax{_PHONE_SEQ[0]}", status="active") + db.add(u) + db.flush() + acc = wallet_repo.get_or_create_account(db, u.id, commit=False) + acc.coin_balance = coin + acc.total_coin_earned = coin + db.flush() + return u.id + + +def test_first_run_converts_and_records_marker() -> None: + """首轮(标记为空)→ 到分全额兑 + 落库当天标记。""" + db = SessionLocal() + try: + uid = _new_user(db, coin=300) + db.commit() + + today = date(2026, 8, 6) + result = w._exchange_if_due(db, today) + + assert result is not None and result["converted"] >= 1 + acc = db.get(CoinAccount, uid) + assert acc.coin_balance == 0 and acc.cash_balance_cents > 0 # 300 为整分,无零头 + assert w._read_last_run(db) == today # 标记落库 + finally: + db.rollback() + db.close() + + +def test_same_day_restart_does_not_resweep() -> None: + """路径 B 回归:0 点兑过后,同日进程重启(下午部署)不得再补扫当天新达标用户。""" + db = SessionLocal() + try: + today = date(2026, 8, 6) + first = _new_user(db, coin=200) + db.commit() + assert w._exchange_if_due(db, today) is not None + assert db.get(CoinAccount, first).coin_balance == 0 # 首轮兑掉 + + # 0 点后才达标的用户(白天攒够;或 0 点是零头、白天赚够) + late = _new_user(db, coin=500) + db.commit() + + # 模拟同一北京日内进程重启 → 持久化标记 == today → 整轮跳过,不碰 late + assert w._exchange_if_due(db, today) is None + acc = db.get(CoinAccount, late) + assert acc.coin_balance == 500 and acc.cash_balance_cents == 0 + assert db.execute(select(CashTransaction).where( + CashTransaction.user_id == late, + CashTransaction.biz_type == "exchange_in", + )).first() is None # late 无任何兑现金流水 + finally: + db.rollback() + db.close() + + +def test_new_day_runs_again() -> None: + """跨到北京新的一天(或真漏了 0 点,标记 < today)→ 照常补跑。""" + db = SessionLocal() + try: + w._write_last_run(db, date(2026, 8, 6)) # 昨天已兑 + late = _new_user(db, coin=500) + db.commit() + + result = w._exchange_if_due(db, date(2026, 8, 7)) + assert result is not None and result["converted"] >= 1 + assert db.get(CoinAccount, late).coin_balance == 0 + assert w._read_last_run(db) == date(2026, 8, 7) + finally: + db.rollback() + db.close() -- 2.52.0