import { id, qualified } from './core.mjs'; function sourceChanged() { return Object.assign(new Error('源图片在任务建立后发生变化或已删除,请检查该批文件'), { code: 'MIGRATION_SOURCE_CHANGED' }); } /** * Validate a bounded batch immediately after the caller begins an InnoDB * REPEATABLE READ transaction, BEFORE any consistent read or explicitly created * consistent snapshot. The plain metadata read must establish its snapshot * only after the primary-key range has been locked. * * The first locking read scans from the minimum expected ID through the upper * key range, including deleted rows and potential newer versions. This retains * range/gap protection until commit/rollback, including against undeleting or * changing another row into a newer matching version. Do not restrict this * range lock to only expected IDs, active rows, or matching paths. * * The caller owns transaction boundaries, rollback on failure and batch size. */ export async function verifySourceBatch(db, config, items) { if (!Array.isArray(items) || items.length === 0) throw new TypeError('源文件校验批次不能为空'); if (!config?.mysql?.database) throw new TypeError('源文件校验需要数据库名称'); const expected = new Map(); for (const item of items) { const contentId = id(item?.contentId); const configId = id(item?.configId); if (expected.has(contentId)) throw new TypeError('源文件校验批次不能包含重复的内容编号'); if (typeof item.path !== 'string' || !item.path.length) throw new TypeError('源文件路径必须为非空字符串'); if (!Number.isSafeInteger(item.size) || item.size < 0 || typeof item.sha256 !== 'string' || !item.sha256.length) { throw new TypeError('源文件校验需要有效的预期大小和 SHA-256'); } expected.set(contentId, { ...item, contentId, configId }); } const entries = [...expected.values()]; const minimumId = entries.reduce((minimum, item) => BigInt(item.contentId) < BigInt(minimum) ? item.contentId : minimum, entries[0].contentId); const sourceTable = qualified(config.mysql.database, 'infra_file_content'); const cases = entries.map(() => 'WHEN config_id=? AND path=? THEN ?').join(' '); const predicates = entries.map(() => '(config_id=? AND path=? AND id>=?)').join(' OR '); const caseParams = entries.flatMap(item => [item.configId, item.path, item.contentId]); const predicateParams = entries.flatMap(item => [item.configId, item.path, item.contentId]); // A primary-key-only locking scan avoids the large-row access path triggered // by locking reads that evaluate config/path predicates. Do not filter out // deleted rows: their undelete could otherwise create a new active version. await db.execute( `SELECT id FROM ${sourceTable} WHERE id>=? ORDER BY id FOR UPDATE`, [minimumId] ); // CASE is evaluated by MySQL using the path column's collation. The returned // expected ID is the only grouping key used in JavaScript. This consistent // read establishes a fresh snapshot while the whole relevant range is held. const [latestRows] = await db.execute( `SELECT id,CASE ${cases} END AS expected_content_id FROM ${sourceTable} WHERE deleted=0 AND id>=? AND (${predicates}) ORDER BY id DESC`, [...caseParams, minimumId, ...predicateParams] ); const latest = new Map(); for (const row of latestRows) { const expectedId = String(row.expected_content_id); if (!expected.has(expectedId)) throw sourceChanged(); const actualId = id(row.id); if (!latest.has(expectedId) || BigInt(actualId) > BigInt(latest.get(expectedId))) latest.set(expectedId, actualId); } for (const contentId of expected.keys()) { if (latest.get(contentId) !== contentId) throw sourceChanged(); } // The range is already locked and each config/path version is validated. // Digest work now touches only the selected primary keys, not a sorted scan. const [contentRows] = await db.execute( `SELECT id,config_id,path,OCTET_LENGTH(content) AS size,LOWER(SHA2(content,256)) AS sha256 FROM ${sourceTable} WHERE deleted=0 AND id IN (${entries.map(() => '?').join(',')}) FOR UPDATE`, entries.map(item => item.contentId) ); const verified = new Set(); for (const row of contentRows) { const contentId = id(row.id); const item = expected.get(contentId); if (!item || verified.has(contentId) || Number(row.size) !== item.size || row.sha256 !== item.sha256) throw sourceChanged(); verified.add(contentId); } if (verified.size !== expected.size) throw sourceChanged(); }