From 2fcd6d996c77576e4ee1df7071e45224026f7425 Mon Sep 17 00:00:00 2001 From: zyc <1439655764@qq.com> Date: Tue, 7 Jul 2026 11:25:40 +0800 Subject: [PATCH] =?UTF-8?q?fix(deploy):=20migrate=20=E8=B7=A8=20pod=20?= =?UTF-8?q?=E4=B8=B2=E8=A1=8C=E5=8C=96=20+=20rescale=20=E8=BF=81=E7=A7=BB?= =?UTF-8?q?=E5=B9=82=E7=AD=89=E5=AE=88=E5=8D=AB(=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E7=8E=AF=E5=A2=83=C3=9710=E2=81=B6=E4=BA=8B=E6=95=85=E5=A4=8D?= =?UTF-8?q?=E7=9B=98)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 事故:连推 3 commit 触发 3 轮 rolling 部署,多个新 pod 并发跑 migrate 且崩溃重跑, 积分 ×10 rescale 被交错重放——ai.0025 执行 6 次(单价/任务计价 ×10⁶),accounts.0008 执行 2 次(限额 ×100),billing.0003/0004 从未完成(django_migrations 漏记录)。 测试库数据已按精确倍率手工修复并补记迁移记录(备份于本机)。 两层防复发: 1. docker-entrypoint 用 MySQL GET_LOCK('airshelf_migrate') 串行化 migrate, 后到 pod 等锁,拿到时迁移已被记录 → 自然 no-op; 2. 三个 rescale 迁移加 airshelf_rescale_marker 幂等标记(与数据变更同事务提交): 记录丢失/崩溃重跑时,标记在 → 跳过,不会重复 ×10。 Co-Authored-By: Claude Fable 5 --- .../migrations/0008_points_rescale.py | 37 +++++++++++++++++-- .../apps/ai/migrations/0025_points_rescale.py | 37 +++++++++++++++++-- .../billing/migrations/0004_points_rescale.py | 37 +++++++++++++++++-- core/backend/docker-entrypoint.sh | 33 ++++++++++++++++- 4 files changed, 133 insertions(+), 11 deletions(-) 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 ;;