diff --git a/core/backend/apps/accounts/migrations/0008_points_rescale.py b/core/backend/apps/accounts/migrations/0008_points_rescale.py index fb64f1c..d91b9cb 100644 --- a/core/backend/apps/accounts/migrations/0008_points_rescale.py +++ b/core/backend/apps/accounts/migrations/0008_points_rescale.py @@ -23,10 +23,41 @@ REVERSE_STATEMENTS = [ ] +# 幂等守卫(2026-07-07 测试环境事故复盘):并发/崩溃后的 migrate 重跑会把 ×10 重复应用 +# (事故中本迁移被交错重放,单价被乘到 ×10⁶)。标记行与数据变更同一事务提交: +# 上次成功 → 标记在 → 跳过;上次崩溃回滚 → 标记不在 → 安全重放。 +MARKER = "accounts.0008_points_rescale" +_MARKER_DDL = "CREATE TABLE IF NOT EXISTS airshelf_rescale_marker (name varchar(80) NOT NULL PRIMARY KEY)" + + def _run(statements): def apply(apps, schema_editor): - with transaction.atomic(using=schema_editor.connection.alias): - with schema_editor.connection.cursor() as cursor: + conn = schema_editor.connection + with conn.cursor() as cursor: # DDL 幂等,MySQL 隐式提交故放事务外 + cursor.execute(_MARKER_DDL) + with transaction.atomic(using=conn.alias): + with conn.cursor() as cursor: + cursor.execute("SELECT COUNT(*) FROM airshelf_rescale_marker WHERE name = %s", [MARKER]) + if cursor.fetchone()[0]: + return # 已应用过(django_migrations 记录丢失/并发重跑),幂等跳过 + cursor.execute("INSERT INTO airshelf_rescale_marker (name) VALUES (%s)", [MARKER]) + for sql in statements: + cursor.execute(sql) + + return apply + + +def _run_reverse(statements): + def apply(apps, schema_editor): + conn = schema_editor.connection + with conn.cursor() as cursor: + cursor.execute(_MARKER_DDL) + with transaction.atomic(using=conn.alias): + with conn.cursor() as cursor: + cursor.execute("SELECT COUNT(*) FROM airshelf_rescale_marker WHERE name = %s", [MARKER]) + if not cursor.fetchone()[0]: + return # 未应用过或已回滚,无需反向 + cursor.execute("DELETE FROM airshelf_rescale_marker WHERE name = %s", [MARKER]) for sql in statements: cursor.execute(sql) @@ -35,4 +66,4 @@ def _run(statements): class Migration(migrations.Migration): dependencies = [("accounts", "0007_team_monthly_credit_limit")] - operations = [migrations.RunPython(_run(RESCALE_STATEMENTS), _run(REVERSE_STATEMENTS))] + operations = [migrations.RunPython(_run(RESCALE_STATEMENTS), _run_reverse(REVERSE_STATEMENTS))] diff --git a/core/backend/apps/ai/migrations/0025_points_rescale.py b/core/backend/apps/ai/migrations/0025_points_rescale.py index 254c580..1a75653 100644 --- a/core/backend/apps/ai/migrations/0025_points_rescale.py +++ b/core/backend/apps/ai/migrations/0025_points_rescale.py @@ -20,10 +20,41 @@ REVERSE_STATEMENTS = [ ] +# 幂等守卫(2026-07-07 测试环境事故复盘):并发/崩溃后的 migrate 重跑会把 ×10 重复应用 +# (事故中本迁移被交错重放,单价被乘到 ×10⁶)。标记行与数据变更同一事务提交: +# 上次成功 → 标记在 → 跳过;上次崩溃回滚 → 标记不在 → 安全重放。 +MARKER = "ai.0025_points_rescale" +_MARKER_DDL = "CREATE TABLE IF NOT EXISTS airshelf_rescale_marker (name varchar(80) NOT NULL PRIMARY KEY)" + + def _run(statements): def apply(apps, schema_editor): - with transaction.atomic(using=schema_editor.connection.alias): - with schema_editor.connection.cursor() as cursor: + conn = schema_editor.connection + with conn.cursor() as cursor: # DDL 幂等,MySQL 隐式提交故放事务外 + cursor.execute(_MARKER_DDL) + with transaction.atomic(using=conn.alias): + with conn.cursor() as cursor: + cursor.execute("SELECT COUNT(*) FROM airshelf_rescale_marker WHERE name = %s", [MARKER]) + if cursor.fetchone()[0]: + return # 已应用过(django_migrations 记录丢失/并发重跑),幂等跳过 + cursor.execute("INSERT INTO airshelf_rescale_marker (name) VALUES (%s)", [MARKER]) + for sql in statements: + cursor.execute(sql) + + return apply + + +def _run_reverse(statements): + def apply(apps, schema_editor): + conn = schema_editor.connection + with conn.cursor() as cursor: + cursor.execute(_MARKER_DDL) + with transaction.atomic(using=conn.alias): + with conn.cursor() as cursor: + cursor.execute("SELECT COUNT(*) FROM airshelf_rescale_marker WHERE name = %s", [MARKER]) + if not cursor.fetchone()[0]: + return # 未应用过或已回滚,无需反向 + cursor.execute("DELETE FROM airshelf_rescale_marker WHERE name = %s", [MARKER]) for sql in statements: cursor.execute(sql) @@ -32,4 +63,4 @@ def _run(statements): class Migration(migrations.Migration): dependencies = [("ai", "0024_aitask_base_cost")] - operations = [migrations.RunPython(_run(RESCALE_STATEMENTS), _run(REVERSE_STATEMENTS))] + operations = [migrations.RunPython(_run(RESCALE_STATEMENTS), _run_reverse(REVERSE_STATEMENTS))] diff --git a/core/backend/apps/billing/migrations/0004_points_rescale.py b/core/backend/apps/billing/migrations/0004_points_rescale.py index c395239..0d1393b 100644 --- a/core/backend/apps/billing/migrations/0004_points_rescale.py +++ b/core/backend/apps/billing/migrations/0004_points_rescale.py @@ -29,10 +29,41 @@ REVERSE_STATEMENTS = [ ] +# 幂等守卫(2026-07-07 测试环境事故复盘):并发/崩溃后的 migrate 重跑会把 ×10 重复应用 +# (事故中本迁移被交错重放,单价被乘到 ×10⁶)。标记行与数据变更同一事务提交: +# 上次成功 → 标记在 → 跳过;上次崩溃回滚 → 标记不在 → 安全重放。 +MARKER = "billing.0004_points_rescale" +_MARKER_DDL = "CREATE TABLE IF NOT EXISTS airshelf_rescale_marker (name varchar(80) NOT NULL PRIMARY KEY)" + + def _run(statements): def apply(apps, schema_editor): - with transaction.atomic(using=schema_editor.connection.alias): - with schema_editor.connection.cursor() as cursor: + conn = schema_editor.connection + with conn.cursor() as cursor: # DDL 幂等,MySQL 隐式提交故放事务外 + cursor.execute(_MARKER_DDL) + with transaction.atomic(using=conn.alias): + with conn.cursor() as cursor: + cursor.execute("SELECT COUNT(*) FROM airshelf_rescale_marker WHERE name = %s", [MARKER]) + if cursor.fetchone()[0]: + return # 已应用过(django_migrations 记录丢失/并发重跑),幂等跳过 + cursor.execute("INSERT INTO airshelf_rescale_marker (name) VALUES (%s)", [MARKER]) + for sql in statements: + cursor.execute(sql) + + return apply + + +def _run_reverse(statements): + def apply(apps, schema_editor): + conn = schema_editor.connection + with conn.cursor() as cursor: + cursor.execute(_MARKER_DDL) + with transaction.atomic(using=conn.alias): + with conn.cursor() as cursor: + cursor.execute("SELECT COUNT(*) FROM airshelf_rescale_marker WHERE name = %s", [MARKER]) + if not cursor.fetchone()[0]: + return # 未应用过或已回滚,无需反向 + cursor.execute("DELETE FROM airshelf_rescale_marker WHERE name = %s", [MARKER]) for sql in statements: cursor.execute(sql) @@ -41,4 +72,4 @@ def _run(statements): class Migration(migrations.Migration): dependencies = [("billing", "0003_billingconfig")] - operations = [migrations.RunPython(_run(RESCALE_STATEMENTS), _run(REVERSE_STATEMENTS))] + operations = [migrations.RunPython(_run(RESCALE_STATEMENTS), _run_reverse(REVERSE_STATEMENTS))] diff --git a/core/backend/docker-entrypoint.sh b/core/backend/docker-entrypoint.sh index 6f05cc9..e3c5375 100644 --- a/core/backend/docker-entrypoint.sh +++ b/core/backend/docker-entrypoint.sh @@ -3,10 +3,39 @@ set -e # Only the web (gunicorn) container should run migrations / collectstatic. # The celery worker shares this image but skips DB schema mutation to avoid races. +# +# ⚠️ migrate 必须跨 pod 串行化(MySQL GET_LOCK):2026-07-07 事故——连推 3 个 commit 触发 +# 3 轮 rolling 部署,多个新 pod 并发跑 migrate,数据迁移(积分 ×10 rescale)被交错重放 6 次, +# 测试库单价被乘成 ×10⁶(¥2 → 2,000,000)。锁把并发压成串行;后到者拿到锁时迁移已被 +# 先到者记录,migrate 自然 no-op。锁超时 600s 拿不到 → 快速失败重启,绝不裸跑。 case "$1" in gunicorn) - echo "[entrypoint] running migrations..." - python manage.py migrate --noinput + echo "[entrypoint] running migrations (serialized via DB advisory lock)..." + python - <<'PYEOF' +import os + +import django + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "airshelf.settings.production") +django.setup() + +from django.core.management import call_command +from django.db import connection + +if connection.vendor == "mysql": + with connection.cursor() as cursor: + cursor.execute("SELECT GET_LOCK('airshelf_migrate', 600)") + acquired = cursor.fetchone()[0] + if not acquired: + raise SystemExit("[entrypoint] FATAL: migrate advisory lock timeout (another pod stuck?)") + try: + call_command("migrate", interactive=False) + finally: + with connection.cursor() as cursor: + cursor.execute("SELECT RELEASE_LOCK('airshelf_migrate')") +else: + call_command("migrate", interactive=False) +PYEOF echo "[entrypoint] collecting static..." python manage.py collectstatic --noinput ;;