yumoqing/pipeline-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(deploy): dmig 台账两个bug——skipped批次误判待执行+detail含引号写入失败

Browse Source 
- ledger_state: applied+skipped 都算已处理，含 failed 的可重试
- _esc/_ledger_write: detail 转义单引号+ON DUPLICATE KEY UPDATE
This commit is contained in:
yumoqing 2026-09-01 16:04:38 +08:00
parent dd40df6337
commit 27b6a7f543
1 changed files with 41 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
deploy/dmig.py
View File

@ -101,16 +101,24 @@ def ensure_ledger(conf):
def ledger_state(conf):
out, err = mysql(conf, "SELECT batch, MAX(step) FROM %s WHERE status='applied' "
"""已处理批次：{batch: (max_step, has_failed)}。applied+skipped 都算已处理，
含 failed 记录的批次不算完成。"""
out, err = mysql(conf, "SELECT batch, MAX(step), SUM(status='failed') FROM %s "
"WHERE status IN ('applied','skipped','failed') "
"GROUP BY batch" % LEDGER_TABLE)
applied = {}
for ln in (out or "").splitlines():
p = ln.split("\t")
if len(p) == 2:
applied[p[0]] = int(p[1])
if len(p) == 3:
applied[p[0]] = (int(p[1]), int(p[2]) > 0)
return applied
def _esc(s):
"""转义 SQL 单引号（detail/backup_file 入台账前必用）。"""
return str(s).replace("\\", "\\\\").replace("'", "''")
def load_migrations():
migs = {}
if not os.path.isdir(MIG_DIR):
@ -183,16 +191,38 @@ def cmd_status(conf):
print("=== 迁移台账 ===")
for mid in sorted(migs):
m = migs[mid]
st = "applied(步骤%d)" % applied[mid] if mid in applied else "待执行"
print(" %s %-10s %s" % (mid, st, m.get("title", "")[:50]))
print("\n待执行批次数:", len([m for m in migs if m not in applied]))
if mid in applied:
mx, failed = applied[mid]
st = "失败(步骤≤%d)" % mx if failed else "已处理(步骤≤%d)" % mx
else:
st = "待执行"
print(" %s %-16s %s" % (mid, st, m.get("title", "")[:50]))
pending = [mid for mid in migs
if mid not in applied or applied[mid][1]]
print("\n待执行批次数:", len(pending), ("(%s)" % ",".join(pending)) if pending else "")
def _pending(migs, applied):
"""未完成的批次：未执行过，或含 failed 记录（修复后可重试）。"""
return [b for b in sorted(migs) if b not in applied or applied[b][1]]
def _ledger_write(conf, mid, step, status, detail, backup_file=""):
_, err = mysql(conf,
"INSERT INTO %s (id, batch, step, status, backup_file, detail) "
"VALUES ('%s', '%s', %d, '%s', '%s', '%s') "
"ON DUPLICATE KEY UPDATE status='%s', detail='%s'"
% (LEDGER_TABLE, _new_id(), mid, step, status,
_esc(backup_file[:200]), _esc(detail[:200]), status, _esc(detail[:200])))
if err:
print(" [WARN] 台账写入失败(不影响库变更): %s" % err[:150])
def cmd_plan(conf, batches):
ensure_ledger(conf)
applied = ledger_state(conf)
migs = load_migrations()
targets = [b for b in (batches or sorted(migs)) if b not in applied]
targets = [b for b in (batches or _pending(migs, applied)) if b not in applied or applied[b][1]]
if not targets:
print("无待执行批次")
return
@ -215,7 +245,7 @@ def cmd_apply(conf, batches):
ensure_ledger(conf)
applied = ledger_state(conf)
migs = load_migrations()
targets = [b for b in (batches or sorted(migs)) if b not in applied]
targets = [b for b in (batches or _pending(migs, applied)) if b not in applied or applied[b][1]]
if not targets:
print("无待执行批次")
return
@ -237,24 +267,17 @@ def cmd_apply(conf, batches):
for i, step in enumerate(m.get("up", []), 1):
sqls, skip = render_step(conf, step, "up")
if skip:
mysql(conf, "INSERT INTO %s (id, batch, step, status, detail) "
"VALUES ('%s', '%s', %d, 'skipped', '%s') "
"ON DUPLICATE KEY UPDATE status='skipped'"
% (LEDGER_TABLE, _new_id(), mid, i, skip[:200]))
_ledger_write(conf, mid, i, "skipped", skip)
print(" 步骤%d 跳过(%s)" % (i, skip))
continue
for s in sqls:
out, err = mysql(conf, s)
if err:
mysql(conf, "INSERT INTO %s (id, batch, step, status, detail) "
"VALUES ('%s', '%s', %d, 'failed', '%s')"
% (LEDGER_TABLE, _new_id(), mid, i, err[:200]))
_ledger_write(conf, mid, i, "failed", err)
die("步骤%d 失败: %s\n（已完成步骤可用 `dmig.py rollback %s` 回退）"
% (i, err, mid))
print(" 步骤%d OK: %s" % (i, s[:90]))
mysql(conf, "INSERT INTO %s (id, batch, step, status, backup_file, detail) "
"VALUES ('%s', '%s', %d, 'applied', '%s', '%s')"
% (LEDGER_TABLE, _new_id(), mid, i, bk[:200], str(step)[:200]))
_ledger_write(conf, mid, i, "applied", str(step), bk)
print(" ✅ %s 完成" % mid)
print("\n全部完成。验证: ./py3/bin/python deploy/dmig.py status && ddiff.py --checklist")