#!/bin/bash
# 盈尺（Surinch）MySQL 逻辑备份恢复演练：8 类恢复失败复现 + RTO 分步计时 v1.0
# 只能在测试实例上跑：会在源实例和目标实例上 DROP/CREATE ${DB}、rst2_lab、blob_lab、wlog_lab，
# 会在目标实例上执行 RESET MASTER、临时改 innodb_flush_log_at_trx_commit / sync_binlog / max_allowed_packet、
# 临时 ALTER INSTANCE DISABLE INNODB REDO_LOG，并建用户 app@'%'、restorer@'%'。目标实例必须是一次性的空实例。
# 测试：MySQL 8.0.46（官方 minimal 包），GTID ON、ROW、sync_binlog=1、innodb_flush_log_at_trx_commit=1、buffer pool 1G；
#       源实例 3306、目标实例 3307（--initialize-insecure 新建），自造数据 orders 200 万 + payments 50 万 + notes 2000，非客户数据。
# 用法：SRC_CNF=/path/src.cnf DST_CNF=/path/dst.cnf WORK=/tmp/rst ./mysql-restore-fail-rto-lab-v1.0.sh | tee result.txt
set -uo pipefail
BIN=${BIN:-/opt/mysql/bin}
SRC_CNF=${SRC_CNF:?需要源测试实例的 --defaults-file}
DST_CNF=${DST_CNF:?需要目标测试实例的 --defaults-file}
DB=${DB:-rst_lab}
WORK=${WORK:-/tmp/rst_lab_work}
case "$DB" in *_lab) ;; *) echo "DB 名必须以 _lab 结尾（防止误删业务库）"; exit 1;; esac
S="$BIN/mysql --defaults-file=$SRC_CNF"
D="$BIN/mysql --defaults-file=$DST_CNF"
DUMP="$BIN/mysqldump --defaults-file=$SRC_CNF"
mkdir -p "$WORK"
ts(){ date '+%F %T.%3N'; }
now(){ date +%s.%N; }
ms(){ awk -v a="$1" -v b="$2" 'BEGIN{printf "%.0f", (b-a)*1000}'; }
sz(){ du -m "$1" | cut -f1; }
# run_restore 标签 客户端命令 库名 文件：计时，打印退出码和前 3 行报错
run_restore(){
  local tag=$1 cli=$2 db=$3 f=$4 t0 t1 rc
  t0=$(now); $cli "$db" < "$f" > "$WORK/$tag.out" 2> "$WORK/$tag.err"; rc=$?; t1=$(now)
  echo "RESULT|$tag|rc=$rc|ms=$(ms $t0 $t1)|err=$(grep -v 'Using a password' "$WORK/$tag.err" | head -3 | tr '\n' ' ')"
}
# 大量写入后紧接着 DROP DATABASE，本机偶发 ERROR 1010（errno 17），稍等重试即可
dropdb(){ local i; for i in 1 2 3 4 5; do $1 -e "DROP DATABASE IF EXISTS $2" 2>/dev/null && return 0; sleep 2; done; echo "WARN|DROP DATABASE $2 失败"; }
dst_reset(){ dropdb "$D" "$DB"; dropdb "$D" rst2_lab; $D -e "RESET MASTER;" 2>&1 | grep -v Warning; }

init(){
  echo "[$(ts)] 源实例建库造数"
  $S <<SQL
DROP DATABASE IF EXISTS $DB; CREATE DATABASE $DB CHARACTER SET utf8mb4; USE $DB;
CREATE USER IF NOT EXISTS 'app'@'%' IDENTIFIED BY 'App_lab_2026!';
GRANT SELECT, INSERT, UPDATE, DELETE, EXECUTE, TRIGGER ON $DB.* TO 'app'@'%';
CREATE TABLE digits(d INT PRIMARY KEY); INSERT INTO digits VALUES (0),(1),(2),(3),(4),(5),(6),(7),(8),(9);
CREATE TABLE orders(
  id BIGINT PRIMARY KEY AUTO_INCREMENT, user_id INT NOT NULL, status TINYINT NOT NULL,
  amount DECIMAL(10,2) NOT NULL, created_at DATETIME NOT NULL, remark VARCHAR(100) NOT NULL,
  KEY idx_user(user_id, created_at), KEY idx_created(created_at)) ENGINE=InnoDB;
INSERT INTO orders(user_id,status,amount,created_at,remark)
SELECT 1+MOD(CRC32(CONCAT('u',n)),100000), 1+MOD(CRC32(CONCAT('s',n)),4),
       ROUND(5+MOD(CRC32(CONCAT('a',n)),20000)/10,2),
       TIMESTAMP('2026-01-01') + INTERVAL MOD(CRC32(CONCAT('d',n)),25000000) SECOND, REPEAT('x',60)
FROM (SELECT a.d+b.d*10+c.d*100+e.d*1000+f.d*10000+g.d*100000+h.d*1000000 n
      FROM digits a,digits b,digits c,digits e,digits f,digits g,(SELECT 0 d UNION ALL SELECT 1) h) t;
CREATE TABLE payments(id BIGINT PRIMARY KEY AUTO_INCREMENT, order_id BIGINT NOT NULL, paid_at DATETIME NOT NULL,
  channel VARCHAR(10) NOT NULL, amount DECIMAL(10,2) NOT NULL, KEY idx_order(order_id)) ENGINE=InnoDB;
INSERT INTO payments(order_id,paid_at,channel,amount) SELECT id, created_at + INTERVAL 5 MINUTE, ELT(1+MOD(id,3),'alipay','wechat','card'), amount FROM orders WHERE MOD(id,4)=0;
CREATE TABLE notes(id INT PRIMARY KEY AUTO_INCREMENT, body VARCHAR(200) NOT NULL) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
INSERT INTO notes(body) SELECT CONCAT('订单备注 ', n, IF(MOD(n,4)=0,' 加急🚀',''), ' café') FROM (SELECT a.d+b.d*10+c.d*100+e.d*1000 n FROM digits a,digits b,digits c,digits e) t WHERE n<2000;
CREATE DEFINER='app'@'%' SQL SECURITY DEFINER VIEW v_paid AS SELECT o.id, o.user_id, p.amount, p.paid_at FROM orders o JOIN payments p ON p.order_id=o.id;
CREATE DEFINER='app'@'%' TRIGGER trg_orders_bi BEFORE INSERT ON orders FOR EACH ROW SET NEW.remark = IF(NEW.remark='', 'n/a', NEW.remark);
CREATE DEFINER='app'@'%' PROCEDURE sp_user_paid(IN u INT) SQL SECURITY DEFINER SELECT COUNT(*) cnt, SUM(amount) total FROM v_paid WHERE user_id=u;
ANALYZE TABLE orders, payments, notes;
SQL
  $S -e "SELECT table_name, table_rows, ROUND((data_length+index_length)/1024/1024,1) total_mb FROM information_schema.tables WHERE table_schema='$DB' ORDER BY 1; SELECT COUNT(*) orders FROM $DB.orders; SELECT COUNT(*) payments FROM $DB.payments; SELECT @@version, @@gtid_executed;"
  $D -e "SELECT @@version, @@port, @@gtid_executed AS dst_gtid_executed;"
}

checksum(){ # 标签 客户端 库
  local t0 t1; t0=$(now)
  echo "CHECKSUM|$1|$($2 -N -e "CHECKSUM TABLE $3.orders, $3.payments, $3.notes" | awk '{printf "%s=%s ", $1, $2}')|ms=$(ms $t0 $(now))"
  t0=$(now); echo "COUNT|$1|$($2 -N -e "SELECT (SELECT COUNT(*) FROM $3.orders),(SELECT COUNT(*) FROM $3.payments),(SELECT COUNT(*) FROM $3.notes)" | tr '\t' '/')|ms=$(ms $t0 $(now))"
}

r1_dump(){
  echo "[$(ts)] R1 全量逻辑备份（--single-transaction --routines --triggers --events，GTID 默认 AUTO）"
  local t0 t1; t0=$(now)
  $DUMP --single-transaction --routines --triggers --events "$DB" > "$WORK/full.sql" 2> "$WORK/full.err"; local rc=$?; t1=$(now)
  echo "RESULT|R1_dump|rc=$rc|ms=$(ms $t0 $t1)|size_mb=$(sz $WORK/full.sql)|warn=$(head -2 $WORK/full.err | tr '\n' ' ')"
  echo "GTIDLINE|$(grep -n 'GTID_PURGED' $WORK/full.sql | head -2 | cut -c1-140 | tr '\n' ' ')|total_lines=$(wc -l < $WORK/full.sql)"
  echo "DEFINER|$(grep -c 'DEFINER=' $WORK/full.sql) 处 DEFINER；$(grep -o 'DEFINER=`[^`]*`@`[^`]*`' $WORK/full.sql | sort | uniq -c | tr '\n' ' ')"
  t0=$(now); gzip -1 -c "$WORK/full.sql" > "$WORK/full.sql.gz"; echo "RESULT|R1_gzip1|ms=$(ms $t0 $(now))|size_mb=$(sz $WORK/full.sql.gz)"
  t0=$(now); cp "$WORK/full.sql" "$WORK/full.copy.sql"; sync; echo "RESULT|R1_copy|ms=$(ms $t0 $(now))"; rm -f "$WORK/full.copy.sql"
  checksum SRC "$S" "$DB"
}

f1_gtid_overlap(){
  echo "[$(ts)] F1 把带 GTID_PURGED 的备份导回非空实例（源实例自己，换库名 rst2_lab）"
  dropdb "$S" rst2_lab; $S -e "CREATE DATABASE rst2_lab"
  run_restore F1_same_instance "$S" rst2_lab "$WORK/full.sql"
  echo "F1_after|tables_in_rst2_lab=$($S -N -e "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='rst2_lab'")"
  dropdb "$S" rst2_lab
}

r2_restore_default(){
  echo "[$(ts)] R2 恢复到全新空实例（默认 sync_binlog=1, innodb_flush_log_at_trx_commit=1）"
  # 目标实例要模拟“新机器”：先删掉上一轮演练留下的 app 用户，否则 F3 复现不出来
  dst_reset; $D -e "DROP USER IF EXISTS 'app'@'%'; CREATE DATABASE $DB"
  $D -e "SELECT @@sync_binlog, @@innodb_flush_log_at_trx_commit, @@gtid_executed"
  run_restore R2_restore_default "$D" "$DB" "$WORK/full.sql"
  $D -N -e "SELECT CONCAT('R2_gtid_after|', @@gtid_executed)"
  checksum DST "$D" "$DB"
  local t0; t0=$(now); $D -e "ANALYZE TABLE $DB.orders, $DB.payments, $DB.notes" >/dev/null; echo "RESULT|R2_analyze|ms=$(ms $t0 $(now))"
  echo "[$(ts)] F3 DEFINER 用户在新实例上不存在"
  $D -e "SHOW WARNINGS" >/dev/null
  $D "$DB" -e "SELECT COUNT(*) FROM v_paid" 2>&1 | tail -1 | sed 's/^/F3_view|/'
  $D "$DB" -e "INSERT INTO orders(user_id,status,amount,created_at,remark) VALUES (1,1,9.9,NOW(),'')" 2>&1 | tail -1 | sed 's/^/F3_insert|/'
  $D "$DB" -e "CALL sp_user_paid(42)" 2>&1 | tail -1 | sed 's/^/F3_call|/'
  echo "[$(ts)] F1b 同一个备份在同一台目标实例上再演练一次（不 RESET MASTER）"
  dropdb "$D" rst2_lab; $D -e "CREATE DATABASE rst2_lab"
  run_restore F1b_second_drill "$D" rst2_lab "$WORK/full.sql"
  dropdb "$D" rst2_lab
  echo "[$(ts)] F3 修复：在目标实例补建 app 用户和授权"
  $D -e "CREATE USER IF NOT EXISTS 'app'@'%' IDENTIFIED BY 'App_lab_2026!'; GRANT SELECT, INSERT, UPDATE, DELETE, EXECUTE ON $DB.* TO 'app'@'%';"
  $D "$DB" -e "INSERT INTO orders(user_id,status,amount,created_at,remark) VALUES (1,1,9.9,NOW(),'')" 2>&1 | tail -1 | sed 's/^/F3_fix1_insert|/'; echo
  echo "[$(ts)] F3 修复第 2 步：按源库 SHOW GRANTS 补齐（少了 TRIGGER 权限）"
  $S -N -e "SHOW GRANTS FOR 'app'@'%'" | sed 's/^/F3_src_grants|/'
  $D -e "GRANT TRIGGER ON $DB.* TO 'app'@'%'"
  $D "$DB" -e "INSERT INTO orders(user_id,status,amount,created_at,remark) VALUES (1,1,9.9,NOW(),'')" 2>&1 | tail -1 | sed 's/^/F3_fix2_insert|/'; echo
  $D "$DB" -N -e "SELECT remark FROM orders ORDER BY id DESC LIMIT 1; CALL sp_user_paid(42)" 2>&1 | tr '\n' ' ' | sed 's/^/F3_fix_call|/'; echo
}

r3_restore_relaxed(){
  echo "[$(ts)] R3 恢复期间临时放宽持久化：innodb_flush_log_at_trx_commit=2, sync_binlog=0"
  dst_reset; $D -e "CREATE DATABASE $DB; SET GLOBAL innodb_flush_log_at_trx_commit=2; SET GLOBAL sync_binlog=0;"
  run_restore R3_restore_flush2 "$D" "$DB" "$WORK/full.sql"
  $D -e "SET GLOBAL innodb_flush_log_at_trx_commit=1; SET GLOBAL sync_binlog=1;"
  echo "[$(ts)] R3b 恢复期间临时调大 innodb_redo_log_capacity 100M → 2G（8.0.30+ 可在线改）"
  dst_reset; $D -e "CREATE DATABASE $DB; SET GLOBAL innodb_redo_log_capacity=2147483648;"
  run_restore R3b_restore_redo2g "$D" "$DB" "$WORK/full.sql"
  $D -e "SET GLOBAL innodb_redo_log_capacity=104857600;"
  echo "[$(ts)] R3c 用 --set-gtid-purged=OFF 的备份恢复（文件里没有 SET SQL_LOG_BIN=0，恢复会写 binlog）"
  $DUMP --single-transaction --routines --triggers --set-gtid-purged=OFF "$DB" > "$WORK/nogtid.sql" 2>/dev/null
  echo "R3c_grep|SQL_LOG_BIN in full.sql=$(grep -c 'SQL_LOG_BIN' $WORK/full.sql)|in nogtid.sql=$(grep -c 'SQL_LOG_BIN' $WORK/nogtid.sql)"
  dst_reset; $D -e "CREATE DATABASE $DB"
  run_restore R3c_restore_binlog_on "$D" "$DB" "$WORK/nogtid.sql"
  $D -N -e "SHOW BINARY LOGS" | awk '{s+=$2} END{printf "R3c_binlog_mb|%.0f\n", s/1024/1024}'
  echo "[$(ts)] R4 恢复期间临时关闭 redo log（ALTER INSTANCE DISABLE INNODB REDO_LOG，8.0.21+，仅限可重建的实例）"
  dst_reset; $D -e "CREATE DATABASE $DB; ALTER INSTANCE DISABLE INNODB REDO_LOG;"
  $D -N -e "SHOW GLOBAL STATUS LIKE 'Innodb_redo_log_enabled'" | sed 's/^/R4_status|/'
  run_restore R4_restore_noredo "$D" "$DB" "$WORK/full.sql"
  $D -e "ALTER INSTANCE ENABLE INNODB REDO_LOG;"; $D -N -e "SHOW GLOBAL STATUS LIKE 'Innodb_redo_log_enabled'" | sed 's/^/R4_status_after|/'
  checksum DST_R4 "$D" "$DB"
}

r5_parallel(){
  echo "[$(ts)] R5 按表拆文件、3 个会话并行导入（--set-gtid-purged=OFF）"
  local t0 t1 t
  t0=$(now)
  for t in orders payments notes; do $DUMP --single-transaction --set-gtid-purged=OFF "$DB" "$t" > "$WORK/t_$t.sql" 2>/dev/null & done; wait
  echo "RESULT|R5_dump_parallel|ms=$(ms $t0 $(now))"
  # 注意：按表导出时触发器已经跟着表文件走了，这里只导存储过程，否则会报 ERROR 1359 Trigger already exists
  $DUMP --single-transaction --set-gtid-purged=OFF --no-data --no-create-info --skip-triggers --routines "$DB" > "$WORK/t_objects.sql" 2>/dev/null
  $DUMP --single-transaction --set-gtid-purged=OFF --no-data --skip-triggers "$DB" v_paid > "$WORK/t_view.sql" 2>/dev/null
  dst_reset; $D -e "CREATE DATABASE $DB"
  t0=$(now)
  for t in orders payments notes; do ( s=$(now); $D "$DB" < "$WORK/t_$t.sql"; echo "RESULT|R5_table_$t|rc=$?|ms=$(ms $s $(now))" ) & done; wait
  t1=$(now); echo "RESULT|R5_restore_parallel_tables|ms=$(ms $t0 $t1)"
  run_restore R5_restore_view "$D" "$DB" "$WORK/t_view.sql"
  run_restore R5_restore_objects "$D" "$DB" "$WORK/t_objects.sql"
  checksum DST_R5 "$D" "$DB"
}

f2_definer_privilege(){
  echo "[$(ts)] F2 用没有 SET_USER_ID / SUPER 的账号导入带 DEFINER 的备份"
  dropdb "$D" rst2_lab; $D -e "CREATE DATABASE rst2_lab; DROP USER IF EXISTS 'restorer'@'%'; CREATE USER 'restorer'@'%' IDENTIFIED BY 'Rst_lab_2026!'; GRANT ALL ON rst2_lab.* TO 'restorer'@'%';"
  local port; port=$($D -N -e "SELECT @@port")
  run_restore F2_restorer "$BIN/mysql -h127.0.0.1 -P$port -urestorer -pRst_lab_2026!" rst2_lab "$WORK/nogtid.sql"
  echo "F2_after|$($D -N -e "SELECT GROUP_CONCAT(CONCAT(table_name,':',table_type) ORDER BY table_name) FROM information_schema.tables WHERE table_schema='rst2_lab'")|triggers=$($D -N -e "SELECT COUNT(*) FROM information_schema.triggers WHERE trigger_schema='rst2_lab'")|routines=$($D -N -e "SELECT COUNT(*) FROM information_schema.routines WHERE routine_schema='rst2_lab'")"
  echo "[$(ts)] F2 修复 A：去掉 DEFINER 子句再导（对象变成 restorer 所有）"
  dropdb "$D" rst2_lab; $D -e "CREATE DATABASE rst2_lab"
  sed -E 's/DEFINER=`[^`]+`@`[^`]+`//g' "$WORK/nogtid.sql" > "$WORK/nodefiner.sql"
  run_restore F2_fix_strip "$BIN/mysql -h127.0.0.1 -P$port -urestorer -pRst_lab_2026!" rst2_lab "$WORK/nodefiner.sql"
  echo "F2_fix_strip_definer|$($D -N -e "SELECT CONCAT(trigger_name,'=',definer) FROM information_schema.triggers WHERE trigger_schema='rst2_lab'")"
  echo "[$(ts)] F2 修复 A2：去掉 DEFINER + 临时 log_bin_trust_function_creators=ON"
  dropdb "$D" rst2_lab; $D -e "CREATE DATABASE rst2_lab"
  $D -e "SET GLOBAL log_bin_trust_function_creators=ON; SHOW WARNINGS" 2>&1 | grep -v '^Level' | sed 's/^/F2_trust_warn|/'
  run_restore F2_fix_strip_trust "$BIN/mysql -h127.0.0.1 -P$port -urestorer -pRst_lab_2026!" rst2_lab "$WORK/nodefiner.sql"
  echo "F2_fix_strip_trust_definer|$($D -N -e "SELECT CONCAT(trigger_name,'=',definer) FROM information_schema.triggers WHERE trigger_schema='rst2_lab'")|view=$($D -N -e "SELECT definer FROM information_schema.views WHERE table_schema='rst2_lab'")"
  $D -e "SET GLOBAL log_bin_trust_function_creators=OFF"
  echo "[$(ts)] F2 修复 B：保留 DEFINER，授予 SET_USER_ID（8.0.46 仍可用，8.2+ 改名 SET_ANY_DEFINER）"
  dropdb "$D" rst2_lab; $D -e "CREATE DATABASE rst2_lab; GRANT SET_USER_ID ON *.* TO 'restorer'@'%'; SHOW WARNINGS;" 2>&1 | grep -v '^$' | sed 's/^/F2_grant|/'
  run_restore F2_fix_grant "$BIN/mysql -h127.0.0.1 -P$port -urestorer -pRst_lab_2026!" rst2_lab "$WORK/nogtid.sql"
  echo "F2_fix_grant_definer|$($D -N -e "SELECT CONCAT(trigger_name,'=',definer) FROM information_schema.triggers WHERE trigger_schema='rst2_lab'")"
  dropdb "$D" rst2_lab
}

f4_no_routines(){
  echo "[$(ts)] F4 备份时没加 --routines（mysqldump 默认不导存储过程）"
  $DUMP --single-transaction --set-gtid-purged=OFF "$DB" > "$WORK/noroutines.sql" 2>/dev/null
  echo "F4_grep|PROCEDURE 出现次数=$(grep -c 'PROCEDURE' $WORK/noroutines.sql)|TRIGGER 出现次数=$(grep -c 'TRIGGER' $WORK/noroutines.sql)"
  dropdb "$D" rst2_lab; $D -e "CREATE DATABASE rst2_lab"
  run_restore F4_restore "$D" rst2_lab "$WORK/noroutines.sql"
  $D rst2_lab -e "CALL sp_user_paid(42)" 2>&1 | tail -1 | sed 's/^/F4_call|/'
  dropdb "$D" rst2_lab
}

f5_packet(){
  echo "[$(ts)] F5 单行 8 MB 的 BLOB，目标实例 max_allowed_packet=4M"
  $S -e "DROP DATABASE IF EXISTS blob_lab; CREATE DATABASE blob_lab; CREATE TABLE blob_lab.att(id INT PRIMARY KEY, data LONGBLOB) ENGINE=InnoDB; INSERT INTO blob_lab.att VALUES (1, REPEAT('a', 1024*1024)), (2, RANDOM_BYTES(1024)), (3, REPEAT(UNHEX('ff'), 8*1024*1024));"
  $DUMP --single-transaction --set-gtid-purged=OFF blob_lab > "$WORK/blob.sql" 2>/dev/null
  $DUMP --single-transaction --set-gtid-purged=OFF --hex-blob blob_lab > "$WORK/blob_hex.sql" 2>/dev/null
  echo "F5_dump|plain_mb=$(sz $WORK/blob.sql)|hex_mb=$(sz $WORK/blob_hex.sql)|longest_line_bytes=$(awk '{ if (length($0)>m) m=length($0) } END{print m}' $WORK/blob.sql)"
  $D -e "DROP DATABASE IF EXISTS blob_lab; CREATE DATABASE blob_lab; SET GLOBAL max_allowed_packet=4194304;"
  run_restore F5_packet_4M "$D" blob_lab "$WORK/blob.sql"
  echo "F5_after|rows=$($D -N -e "SELECT COUNT(*) FROM blob_lab.att" 2>&1)"
  $D -e "SET GLOBAL max_allowed_packet=67108864; DROP DATABASE blob_lab; CREATE DATABASE blob_lab;"
  run_restore F5_packet_64M "$D" blob_lab "$WORK/blob.sql"
  echo "F5_fix|$($D -N -e "SELECT id, LENGTH(data), MD5(data) FROM blob_lab.att ORDER BY id" | tr '\t\n' ', ')|src=$($S -N -e "SELECT id, LENGTH(data), MD5(data) FROM blob_lab.att ORDER BY id" | tr '\t\n' ', ')"
  $D -e "DROP DATABASE blob_lab"; $S -e "DROP DATABASE blob_lab"
}

f6_charset(){
  echo "[$(ts)] F6 用 --default-character-set=latin1 备份 utf8mb4 数据"
  $DUMP --single-transaction --set-gtid-purged=OFF --default-character-set=latin1 "$DB" notes > "$WORK/notes_latin1.sql" 2> "$WORK/notes_latin1.err"
  echo "F6_dump_rc=$?|$(head -2 $WORK/notes_latin1.err | tr '\n' ' ')"
  dropdb "$D" rst2_lab; $D -e "CREATE DATABASE rst2_lab CHARACTER SET utf8mb4"
  run_restore F6_restore "$D" rst2_lab "$WORK/notes_latin1.sql"
  echo "F6_count|src=$($S -N -e "SELECT COUNT(*) FROM $DB.notes")|dst=$($D -N -e "SELECT COUNT(*) FROM rst2_lab.notes")"
  echo "F6_checksum|src=$($S -N -e "CHECKSUM TABLE $DB.notes" | cut -f2)|dst=$($D -N -e "CHECKSUM TABLE rst2_lab.notes" | cut -f2)"
  echo "F6_sample|src=$($S -N -e "SELECT body FROM $DB.notes WHERE id=5")|dst=$($D -N -e "SELECT body FROM rst2_lab.notes WHERE id=5")"
  echo "F6_bad_rows|$($D -N -e "SELECT COUNT(*) FROM rst2_lab.notes WHERE body LIKE '%?%'")|emoji_rows_src=$($S -N -e "SELECT COUNT(*) FROM $DB.notes WHERE body LIKE '%🚀%'")"
  dropdb "$D" rst2_lab
}

writer(){ # 在 wlog_lab 记录每次 INSERT 的耗时（微秒）
  dropdb "$S" wlog_lab
  $S <<SQL 2>&1 | grep -v Warning
CREATE DATABASE wlog_lab; CREATE TABLE wlog_lab.lat(id INT AUTO_INCREMENT PRIMARY KEY, tag VARCHAR(20), us BIGINT, at DATETIME(6)) ENGINE=InnoDB;
DELIMITER //
CREATE PROCEDURE wlog_lab.w(IN tg VARCHAR(20), IN n INT) BEGIN DECLARE i INT DEFAULT 0; DECLARE t0 DATETIME(6);
  WHILE i < n DO SET t0 = SYSDATE(6);
    INSERT INTO $DB.orders(user_id,status,amount,created_at,remark) VALUES (7,1,1.00,NOW(),'w');
    INSERT INTO wlog_lab.lat(tag,us,at) VALUES (tg, TIMESTAMPDIFF(MICROSECOND, t0, SYSDATE(6)), SYSDATE(6));
    DO SLEEP(0.01); SET i = i + 1; END WHILE; END //
DELIMITER ;
SQL
}

f7_lock_tables(){
  echo "[$(ts)] F7 备份时不加 --single-transaction（默认 --lock-tables），同时有写入"
  writer
  local mode t0
  for mode in lock_tables single_trx; do
    $S -e "CALL wlog_lab.w('$mode', 100000)" 2>/dev/null & local wp=$!
    sleep 2; t0=$(now)
    if [ $mode = lock_tables ]; then $DUMP --set-gtid-purged=OFF "$DB" > "$WORK/f7_$mode.sql" 2>/dev/null
    else $DUMP --single-transaction --set-gtid-purged=OFF "$DB" > "$WORK/f7_$mode.sql" 2>/dev/null; fi
    echo "RESULT|F7_dump_$mode|ms=$(ms $t0 $(now))"
    sleep 2; kill $wp 2>/dev/null; $S -N -e "SELECT id FROM information_schema.processlist WHERE info LIKE 'CALL wlog_lab.w%'" | while read id; do $S -e "KILL $id"; done; wait $wp 2>/dev/null
    $S -N -e "SELECT CONCAT('F7_lat|$mode|inserts=',COUNT(*),'|max_ms=',ROUND(MAX(us)/1000,1),'|avg_ms=',ROUND(AVG(us)/1000,2),'|over_1s=',SUM(us>1000000)) FROM wlog_lab.lat WHERE tag='$mode'"
  done
  dropdb "$S" wlog_lab; $S -e "DELETE FROM $DB.orders WHERE remark='w'"
}

f8_ddl_during_dump(){
  echo "[$(ts)] F8 --single-transaction 备份进行中，另一个会话重建了排在后面、还没导到的表（notes）"
  local t0; t0=$(now)
  $DUMP --single-transaction --set-gtid-purged=OFF "$DB" orders payments notes > "$WORK/f8.sql" 2> "$WORK/f8.err" & local dp=$!
  sleep 0.5; local a0; a0=$(now); $S -e "ALTER TABLE $DB.notes ENGINE=InnoDB"; echo "RESULT|F8_alter_notes|ms=$(ms $a0 $(now))"
  wait $dp; local rc=$?
  echo "RESULT|F8_dump|rc=$rc|ms=$(ms $t0 $(now))|err=$(head -2 $WORK/f8.err | tr '\n' ' ')|notes_inserts=$(grep -c 'INSERT INTO `notes`' $WORK/f8.sql)|tail=$(tail -c 100 $WORK/f8.sql | tr '\n' ' ')"
}

STEPS=${STEPS:-"init r1_dump f1_gtid_overlap r2_restore_default r3_restore_relaxed r5_parallel f2_definer_privilege f4_no_routines f5_packet f6_charset f7_lock_tables f8_ddl_during_dump"}
for s in $STEPS; do $s; done
echo "[$(ts)] 结束。清理：DROP DATABASE $DB（源/目标）、DROP USER app@'%', restorer@'%'（目标）需手动执行。"