diff --git a/alembic/versions/guide_video_play_user_seq_unique.py b/alembic/versions/guide_video_play_user_seq_unique.py new file mode 100644 index 0000000..e4c105e --- /dev/null +++ b/alembic/versions/guide_video_play_user_seq_unique.py @@ -0,0 +1,40 @@ +"""guide_video_play 加 (user_id, seq) 唯一约束:堵住并发 /start 绕过次数上限 + +start_play 是无锁 check-then-insert(读 COUNT(*) 算 seq=used+1 再插一行),N 个并发 +/start 会都读到同一个 used、算出同一个 seq、各插一行拿到各自的 play_token,于是 3 次 +上限被绕过、每个 token 都能换 120 金币。加唯一键后并发同 seq 必撞,start_play 捕获 +IntegrityError 降级为 should_play=false(客户端照旧放广告)。 + +用 unique index 而不是 batch_alter_table 加 UniqueConstraint:SQLite 加约束要整表重建, +而 CREATE UNIQUE INDEX 两边都原生支持,回滚也干净。 + +注:若库里已有并发产生的重复 (user_id, seq),建索引会失败 —— 本功能尚未上线,表通常是空的; +真撞上了先按 seq 去重(留 id 最小的一行,多发的金币按 scripts/reset_guide_video.py 的口径退)。 + +Revision ID: guide_video_user_seq_uq +Revises: d9c03cc3ea07 +Create Date: 2026-07-24 10:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "guide_video_user_seq_uq" +down_revision: Union[str, Sequence[str], None] = "d9c03cc3ea07" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_index( + "uq_guide_video_play_user_seq", + "guide_video_play", + ["user_id", "seq"], + unique=True, + ) + + +def downgrade() -> None: + op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play") diff --git a/app/models/guide_video.py b/app/models/guide_video.py index d8774ce..4ecd2a5 100644 --- a/app/models/guide_video.py +++ b/app/models/guide_video.py @@ -8,7 +8,10 @@ - **开播即计数**:客户端每次要展示浮层时调 `/api/v1/guide-video/start`,服务端当场 写一行(status='playing')并返回 play_token;`COUNT(*)` 即已用次数。用户中途 kill App 也算用掉一次(产品选定口径,防反复进出刷金币)。 -- **发币幂等**靠 play_token 唯一键:同一次播放重复上报只发一次。 +- **发币幂等**靠 play_token 定位 + `status='playing'` 条件更新:并发两次上报只有一次 + 改到行(另一次 rowcount=0),所以只发一次币。光有 play_token 唯一键挡不住 —— 发币走的是 + UPDATE,不 INSERT,撞不到任何唯一键。 +- **次数上限**靠 (user_id, seq) 唯一键兜底,防并发 /start 绕过 COUNT 判定(见下)。 与广告收益(ad_feed_reward_record)彻底分离:引导视频不是广告,不该进广告收益报表。 """ @@ -16,7 +19,7 @@ from __future__ import annotations from datetime import datetime -from sqlalchemy import DateTime, ForeignKey, Integer, String, UniqueConstraint, func +from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, UniqueConstraint, func from sqlalchemy.orm import Mapped, mapped_column from app.db.base import Base @@ -29,6 +32,13 @@ class GuideVideoPlay(Base): __table_args__ = ( # 客户端幂等键:同一次播放重复上报奖励只发一次。 UniqueConstraint("play_token", name="uq_guide_video_play_token"), + # 次数上限的**硬约束**:start_play 是无锁 check-then-insert(读 COUNT 算 seq 再插), + # N 个并发 /start 会都读到同一个已用次数、算出同一个 seq,不拦就能各拿一个 token、 + # 各发一次金币,3 次上限形同虚设(改包即可无限刷)。seq 唯一 → 并发同 seq 必撞, + # start_play 捕获 IntegrityError 降级成"这次不放视频"。 + # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束 + # 要整表重建),autogenerate 才不会每次报一条假 diff。 + Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) @@ -39,7 +49,8 @@ class GuideVideoPlay(Base): play_token: Mapped[str] = mapped_column(String(64), nullable=False) # 触发场景:目前只有 coupon(领券等候浮层);留字段以便日后比价等场景复用。 scene: Mapped[str] = mapped_column(String(16), nullable=False, default="coupon") - # 本账号第几次(1-based),= 建行时已有行数 + 1。仅留痕/排查用,判定仍以 COUNT 为准。 + # 本账号第几次(1-based),= 建行时已有行数 + 1。日常判定仍以 COUNT 为准,但 (user_id, seq) + # 唯一键让并发 /start 只能成一个 —— 见 __table_args__。 seq: Mapped[int] = mapped_column(Integer, nullable=False, default=1) # 当次下发的视频地址(运营换片后能回溯用户当时看的是哪支)。 video_url: Mapped[str | None] = mapped_column(String(512), nullable=True) diff --git a/app/repositories/guide_video.py b/app/repositories/guide_video.py index 9ddb6fd..cd30f90 100644 --- a/app/repositories/guide_video.py +++ b/app/repositories/guide_video.py @@ -7,9 +7,13 @@ **计次**按账号(user_id)、**开播即计数**:客户端每次要展示领券等候浮层时调 `/api/v1/guide-video/start`,命中则当场写一行 guide_video_play(status='playing')。 已用次数 = 该账号的行数,达到 max_plays 后不再下发,客户端改放广告(原逻辑)。 +COUNT 判定本身无锁,真正卡住次数上限的是 (user_id, seq) 唯一键:并发 /start 只能成一个。 -**发币**幂等键是 play_token:同一次播放重复上报只入账一次(网络重试 / 关闭与播完同时触发 -都靠它挡住)。中途关闭也照发 —— 产品拍板「中途关闭也算看完」。 +**发币**幂等键是 play_token,落地方式是 `status='playing' → 'granted'` 的**条件更新**: +同一次播放重复上报只入账一次(网络重试 / 关闭与播完同时触发都靠它挡住)。 +中途关闭也照发 —— 产品拍板「中途关闭也算看完」。 + +两处都是直接铸币的路径,改动前先看 `start_play` / `grant_play` 上的并发注释。 """ from __future__ import annotations @@ -17,7 +21,7 @@ import uuid from datetime import datetime from typing import Any -from sqlalchemy import func, select +from sqlalchemy import func, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -169,16 +173,19 @@ def start_play( max_plays = int(cfg.get("max_plays") or 0) reward_coin = int(cfg.get("reward_coin") or 0) used = used_plays(db, user_id) - miss = { - "should_play": False, - "video_url": None, - "play_token": "", - "reward_coin": reward_coin, - "seq": used, - "remaining": max(0, max_plays - used), - } + + def _miss(used_now: int) -> dict[str, Any]: + return { + "should_play": False, + "video_url": None, + "play_token": "", + "reward_coin": reward_coin, + "seq": used_now, + "remaining": max(0, max_plays - used_now), + } + if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays: - return miss + return _miss(used) seq = used + 1 play = GuideVideoPlay( @@ -193,10 +200,17 @@ def start_play( started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None), ) db.add(play) - if commit: - db.commit() - else: - db.flush() + # 上面的 COUNT 判定是无锁 check-then-insert:并发 /start 会都算出同一个 seq。 + # (user_id, seq) 唯一键让只有一个能落库,其余撞键 → 回滚后按"这次不放视频"降级, + # 客户端照旧走广告链路。没有它,并发就能绕过 max_plays 无限刷金币。 + try: + if commit: + db.commit() + else: + db.flush() + except IntegrityError: + db.rollback() + return _miss(used_plays(db, user_id)) return { "should_play": True, "video_url": video_url, @@ -207,6 +221,16 @@ def start_play( } +def _find_play(db: Session, user_id: int, token: str) -> GuideVideoPlay | None: + """按 (play_token, user_id) 取播放行 —— 带 user_id 是防拿别人的 token 来兑。""" + return db.execute( + select(GuideVideoPlay).where( + GuideVideoPlay.play_token == token, + GuideVideoPlay.user_id == user_id, + ) + ).scalar_one_or_none() + + def grant_play( db: Session, user_id: int, *, play_token: str, completed: bool ) -> dict[str, Any]: @@ -216,23 +240,43 @@ def grant_play( 重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)。 """ token = (play_token or "").strip() - play = db.execute( - select(GuideVideoPlay).where( - GuideVideoPlay.play_token == token, - GuideVideoPlay.user_id == user_id, - ) - ).scalar_one_or_none() - if play is None: - return {"granted": False, "coin": 0, "status": "not_found"} - if play.status == "granted": - return {"granted": False, "coin": play.coin, "status": "already_granted"} - # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。 coin = int(get_config(db).get("reward_coin") or 0) - play.completed = 1 if completed else 0 - play.status = "granted" - play.coin = coin - play.granted_at = datetime.now(rewards.CN_TZ).replace(tzinfo=None) + + # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。 + # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing', + # 无锁的话就都往下发币、都 commit,金币入账两次(不用恶意,重试就会中招)。改成条件更新后 + # 并发里只有一条 rowcount=1,另一条拿 0 → 按已发返回,不二次铸币。 + # (PG READ COMMITTED 下后到的 UPDATE 阻塞到对手提交,再按新版本重判 status;SQLite 写串行。) + # + # 别指望 IntegrityError 兜底:这里只 UPDATE 不 INSERT,撞不到 uq_guide_video_play_token; + # 而 biz_type='guide_video' 的金币流水也不在 ux_coin_transaction_task_ref 的谓词 + # (biz_type LIKE 'task%')覆盖范围内 —— 两个唯一键在这条路径上都是不生效的。 + won = db.execute( + update(GuideVideoPlay) + .where( + GuideVideoPlay.play_token == token, + GuideVideoPlay.user_id == user_id, + GuideVideoPlay.status == "playing", + ) + .values( + status="granted", + coin=coin, + completed=1 if completed else 0, + granted_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None), + ) + .execution_options(synchronize_session=False) + ).rowcount + + if not won: + # 没抢到:token 不存在 / 不是本人的 / 已被另一次上报发过。回滚拿干净快照再区分两者 + # (对手此时必然已提交 —— 我们就是被它挡下的,所以读得到它写的 coin)。 + db.rollback() + play = _find_play(db, user_id, token) + if play is None: + return {"granted": False, "coin": 0, "status": "not_found"} + return {"granted": False, "coin": play.coin, "status": "already_granted"} + if coin > 0: crud_wallet.grant_coins( db, @@ -242,16 +286,5 @@ def grant_play( ref_id=token, remark="新手引导视频奖励", ) - try: - db.commit() - except IntegrityError: - # 并发双发(播完 + ✕ 同时到)时其中一条会撞唯一键/行锁,回滚后按已发返回。 - db.rollback() - again = db.execute( - select(GuideVideoPlay).where( - GuideVideoPlay.play_token == token, - GuideVideoPlay.user_id == user_id, - ) - ).scalar_one_or_none() - return {"granted": False, "coin": again.coin if again else 0, "status": "already_granted"} + db.commit() return {"granted": True, "coin": coin, "status": "granted"} diff --git a/tests/test_guide_video.py b/tests/test_guide_video.py new file mode 100644 index 0000000..dfec688 --- /dev/null +++ b/tests/test_guide_video.py @@ -0,0 +1,263 @@ +"""新手引导视频:次数上限 + 发币幂等。 + +这条链路直接铸币,两处并发缺陷曾经都是真漏洞,所以本文件的重点不是走通 happy path, +而是**并发失败分支**: + - `/start` 并发算出同一个 seq → (user_id, seq) 唯一键挡下,降级为"这次不放视频"; + - `/reward` 并发抢跑 → status 条件更新只让一个 rowcount=1,输的那个不得入账。 + +真并发在 SQLite 测试库里复现不了(读事务会直接把写方锁死,而不是让它读到旧快照), +所以用"把对手已经落库的状态先摆好、再让被测方按旧读数往下走"来模拟,断言的是同一件事: +输的一方必须空手而归,且账不能乱。 +""" +from __future__ import annotations + +import pytest +from sqlalchemy import func, select + +from app.db.session import SessionLocal +from app.models.guide_video import GuideVideoPlay +from app.models.wallet import CoinAccount, CoinTransaction +from app.repositories import guide_video as crud_guide +from app.repositories.user import get_user_by_phone + +VIDEO_URL = "/media/guide_video/pytest_guide.mp4" + + +def _login(client, phone: str) -> str: + client.post("/api/v1/auth/sms/send", json={"phone": phone}) + r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"}) + assert r.status_code == 200, r.text + return r.json()["access_token"] + + +def _auth(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +def _user_id(phone: str) -> int: + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + return user.id + + +def _coin_balance(user_id: int) -> int: + with SessionLocal() as db: + acc = db.get(CoinAccount, user_id) + return acc.coin_balance if acc else 0 + + +def _guide_txns(user_id: int) -> list[CoinTransaction]: + """该用户的引导视频金币流水(按 id 升序)。""" + with SessionLocal() as db: + return list(db.execute( + select(CoinTransaction) + .where( + CoinTransaction.user_id == user_id, + CoinTransaction.biz_type == crud_guide.BIZ_TYPE, + ) + .order_by(CoinTransaction.id) + ).scalars().all()) + + +def _assert_ledger_balanced(user_id: int) -> None: + """coin_balance 必须恒等于流水总和 —— 双发/丢更新都会先在这里露馅。""" + with SessionLocal() as db: + total = db.execute( + select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where( + CoinTransaction.user_id == user_id + ) + ).scalar_one() + acc = db.get(CoinAccount, user_id) + assert acc is not None + assert acc.coin_balance == total, f"余额 {acc.coin_balance} != 流水总和 {total}" + + +@pytest.fixture() +def guide_configured(): + """给全局配置塞一支片子(默认 max_plays=3 / reward_coin=120),用完还原成未配片。""" + with SessionLocal() as db: + crud_guide.set_video(db, VIDEO_URL, admin_id=1) + cfg = crud_guide.get_config(db) + yield cfg + with SessionLocal() as db: + crud_guide.set_video(db, None, admin_id=1) + + +# ===== 基本闭环 ===== + + +def test_start_miss_when_no_video_configured(client) -> None: + """没配片 → should_play=False,客户端照旧放广告(本功能不配视频就等于没上线)。""" + token = _login(client, "13920000001") + r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)) + assert r.status_code == 200, r.text + body = r.json() + assert body["should_play"] is False + assert body["play_token"] == "" + + +def test_start_then_reward_grants_once(client, guide_configured) -> None: + """开播 → 上报 → 到账 120;重复上报不再加钱,只回已发金额。""" + cfg = guide_configured + phone = "13920000002" + token = _login(client, phone) + uid = _user_id(phone) + + r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)) + body = r.json() + assert body["should_play"] is True + assert body["video_url"] == VIDEO_URL + assert body["seq"] == 1 + assert body["reward_coin"] == cfg["reward_coin"] + play_token = body["play_token"] + assert play_token + + r = client.post( + "/api/v1/guide-video/reward", + json={"play_token": play_token, "completed": True}, + headers=_auth(token), + ) + assert r.status_code == 200, r.text + assert r.json() == {"granted": True, "coin": cfg["reward_coin"], "status": "granted"} + assert _coin_balance(uid) == cfg["reward_coin"] + + # 重复上报(客户端超时重试 / 播完与 ✕ 都报了一次) + r = client.post( + "/api/v1/guide-video/reward", + json={"play_token": play_token, "completed": False}, + headers=_auth(token), + ) + assert r.json() == {"granted": False, "coin": cfg["reward_coin"], "status": "already_granted"} + assert _coin_balance(uid) == cfg["reward_coin"], "重复上报不得二次入账" + assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水" + _assert_ledger_balanced(uid) + + +def test_reward_rejects_unknown_and_others_token(client, guide_configured) -> None: + """乱填 token / 拿别人的 token 都发不出币(grant 按 user_id + token 双条件定位)。""" + victim = _login(client, "13920000003") + r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(victim)) + stolen = r.json()["play_token"] + + attacker_phone = "13920000004" + attacker = _login(client, attacker_phone) + attacker_uid = _user_id(attacker_phone) + + r = client.post( + "/api/v1/guide-video/reward", + json={"play_token": stolen, "completed": True}, + headers=_auth(attacker), + ) + assert r.json() == {"granted": False, "coin": 0, "status": "not_found"} + + r = client.post( + "/api/v1/guide-video/reward", + json={"play_token": "deadbeef" * 4, "completed": True}, + headers=_auth(attacker), + ) + assert r.json() == {"granted": False, "coin": 0, "status": "not_found"} + assert _coin_balance(attacker_uid) == 0 + + +def test_start_stops_at_max_plays(client, guide_configured) -> None: + """开播即计次:用满 max_plays 后不再下发,客户端回到广告链路。""" + max_plays = int(guide_configured["max_plays"]) + token = _login(client, "13920000005") + + for i in range(1, max_plays + 1): + body = client.post( + "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token) + ).json() + assert body["should_play"] is True, f"第 {i} 次应该还能放" + assert body["seq"] == i + assert body["remaining"] == max_plays - i + + body = client.post( + "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token) + ).json() + assert body["should_play"] is False + assert body["remaining"] == 0 + + +# ===== 并发失败分支(回归) ===== + + +def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypatch) -> None: + """并发 /start 抢到同一个 seq 时,输的一方降级成"不放视频",而不是多拿一个 token。 + + 模拟:对手已经提交了 seq=1,而本次请求读到的还是旧计数(used=0)—— 这正是无锁 + check-then-insert 的race window。没有 (user_id, seq) 唯一键的话,这里会插成第二行、 + 换回第二个 play_token,3 次上限就能被并发无限绕过(改包即可 N 倍刷币)。 + """ + phone = "13920000006" + token = _login(client, phone) + uid = _user_id(phone) + + # 对手先落一行 seq=1 + first = client.post( + "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token) + ).json() + assert first["should_play"] is True and first["seq"] == 1 + + # 本次请求读到的是过期计数 → 仍会算出 seq=1 + monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0) + with SessionLocal() as db: + result = crud_guide.start_play(db, uid) + + assert result["should_play"] is False, "撞 seq 唯一键后必须降级,不能再发一个 token" + assert result["play_token"] == "" + + monkeypatch.undo() + with SessionLocal() as db: + rows = db.execute( + select(func.count()).select_from(GuideVideoPlay).where( + GuideVideoPlay.user_id == uid + ) + ).scalar_one() + assert rows == 1, "抢输的那次不该留下播放行" + + +def test_reward_loses_race_does_not_double_mint(client, guide_configured) -> None: + """并发 /reward 抢输的一方不得入账 —— 条件更新(status 进 WHERE)的失败分支。 + + 模拟的是真并发的**必要条件**:输的一方在对手提交之前就已经读到 status='playing', + 之后带着这个旧认知继续往下走(「播完」与「✕ 关闭」抢跑、或超时重试都会造出这一幕)。 + 这里靠 Session 的 identity map 固化那次旧读数 —— 后续同一 Session 再查同一行,拿回的 + 还是这份旧快照,等价于 PG READ COMMITTED 下两个事务都读到 playing。 + + 旧实现在这里会照发第二笔:它先读再判再写,而 `except IntegrityError` 兜底根本不可能 + 触发 —— 发币走 UPDATE 撞不到 uq_guide_video_play_token,biz_type='guide_video' 的流水 + 也不在 ux_coin_transaction_task_ref 的谓词(biz_type LIKE 'task%')覆盖内。 + """ + coin = int(guide_configured["reward_coin"]) + phone = "13920000007" + token = _login(client, phone) + uid = _user_id(phone) + + play_token = client.post( + "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token) + ).json()["play_token"] + + db_loser = SessionLocal() + try: + # 输的一方先读到 playing —— 并发下两个请求都会读到它 + stale = db_loser.execute( + select(GuideVideoPlay).where(GuideVideoPlay.play_token == play_token) + ).scalar_one() + assert stale.status == "playing" + + # 对手抢先发币并提交 + with SessionLocal() as db_winner: + won = crud_guide.grant_play(db_winner, uid, play_token=play_token, completed=True) + assert won == {"granted": True, "coin": coin, "status": "granted"} + + # 输的一方带着旧认知继续:必须空手而归 + lost = crud_guide.grant_play(db_loser, uid, play_token=play_token, completed=False) + assert lost == {"granted": False, "coin": coin, "status": "already_granted"} + finally: + db_loser.close() + + assert _coin_balance(uid) == coin, "一次播放只能发一次币" + assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水" + _assert_ledger_balanced(uid)