89 lines
4.0 KiB
JavaScript
89 lines
4.0 KiB
JavaScript
import { performance } from 'node:perf_hooks';
|
|
import { qualified, sha256 } from './core.mjs';
|
|
import { applyChanges } from './journal.mjs';
|
|
import { verifySourceBatch } from './source-batch.mjs';
|
|
|
|
/**
|
|
* Switch a verified batch in one caller-session REPEATABLE READ transaction.
|
|
* Source validation must be the first operation after BEGIN: it establishes
|
|
* the source range lock before the transaction's consistent-read snapshot.
|
|
*/
|
|
export async function switchFiles(db, config, items) {
|
|
if (!Array.isArray(items)) throw new TypeError('文件切换批次必须为数组');
|
|
const timings = { sourceMs: 0, writeMs: 0, statusMs: 0, commitMs: 0 };
|
|
if (!items.length) return timings;
|
|
const changes = items.flatMap(item => {
|
|
if (!Array.isArray(item?.files)) throw new TypeError('文件切换项必须包含 files 数组（无文件记录时使用空数组）');
|
|
return item.files.map(file => ({
|
|
schema: config.mysql.database, table: 'infra_file', primaryKey: 'id', pkValue: file.id,
|
|
before: {
|
|
config_id: file.config_id, path: file.path, url: file.url,
|
|
type: file.type, size: file.size, deleted: Buffer.from([0])
|
|
},
|
|
after: {
|
|
config_id: config.targetConfigId, path: item.key, url: item.url,
|
|
type: item.type, size: item.size
|
|
}
|
|
}));
|
|
});
|
|
const keys = items.map(item => sha256(`${item.configId}\0${item.path}`));
|
|
if (new Set(keys).size !== keys.length) throw new TypeError('文件切换批次不能包含重复源路径');
|
|
if (new Set(changes.map(change => String(change.pkValue))).size !== changes.length) {
|
|
throw new TypeError('文件切换批次不能包含重复文件记录');
|
|
}
|
|
const measure = async (name, action) => {
|
|
const started = performance.now();
|
|
try { return await action(); }
|
|
finally { timings[name] += performance.now() - started; }
|
|
};
|
|
|
|
await db.beginTransaction();
|
|
let committing = false;
|
|
try {
|
|
// Do not insert metadata lookups or other consistent reads before this.
|
|
await measure('sourceMs', () => verifySourceBatch(db, config, items));
|
|
await measure('writeMs', async () => {
|
|
// One source path may have many infra_file rows. Bound prepared-statement
|
|
// size while retaining the same source locks and transaction for all rows.
|
|
for (let offset = 0; offset < changes.length; offset += 100) {
|
|
await applyChanges(db, config.mysql.database, config.runId, changes.slice(offset, offset + 100));
|
|
}
|
|
});
|
|
await measure('statusMs', async () => {
|
|
const [result] = await db.execute(
|
|
`UPDATE ${qualified(config.mysql.database, 'infra_file_oss_migration_items')}
|
|
SET status='switched',error=NULL,updated_at=CURRENT_TIMESTAMP
|
|
WHERE run_id=? AND status='verified' AND item_key IN (${keys.map(() => '?').join(',')})`,
|
|
[config.runId, ...keys]);
|
|
if (result.affectedRows !== items.length) {
|
|
throw Object.assign(new Error('部分迁移项已变化或不再处于 verified 状态，必须回滚整批事务'), {
|
|
code: 'MIGRATION_CAS_CONFLICT'
|
|
});
|
|
}
|
|
if (result.warningStatus > 0) {
|
|
throw Object.assign(new Error('批量迁移状态更新产生数据库警告，必须回滚整批事务'), {
|
|
code: 'MIGRATION_SQL_WARNING'
|
|
});
|
|
}
|
|
});
|
|
committing = true;
|
|
await measure('commitMs', () => db.commit());
|
|
return timings;
|
|
} catch (error) {
|
|
try {
|
|
await db.rollback();
|
|
} catch (rollbackError) {
|
|
throw Object.assign(new Error('事务回滚失败，连接状态不确定，禁止降级重试', { cause: error }), {
|
|
code: 'MIGRATION_ROLLBACK_FAILED', fatal: true, noFallback: true, rollbackError
|
|
});
|
|
}
|
|
if (committing) {
|
|
// A failed COMMIT acknowledgement does not prove the server rolled back.
|
|
throw Object.assign(new Error('事务提交结果不确定，禁止降级重试，请重新核对迁移状态', { cause: error }), {
|
|
code: error.code || 'MIGRATION_COMMIT_FAILED', fatal: true, noFallback: true
|
|
});
|
|
}
|
|
throw error;
|
|
}
|
|
}
|