252 lines
12 KiB
JavaScript
252 lines
12 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { switchFiles } from '../lib/transition.mjs';
|
|
import { sha256 } from '../lib/core.mjs';
|
|
|
|
const config = { mysql: { database: 'transition_test' }, runId: 'run', targetConfigId: '20' };
|
|
const itemKey = item => sha256(`${item.configId}\0${item.path}`);
|
|
const clone = value => Buffer.isBuffer(value) ? Buffer.from(value)
|
|
: Array.isArray(value) ? value.map(clone)
|
|
: value && typeof value === 'object' ? Object.fromEntries(Object.entries(value).map(([key, item]) => [key, clone(item)])) : value;
|
|
|
|
function fixture({ orphan = false, count = 1 } = {}) {
|
|
const item = {
|
|
contentId: '9007199254740993', configId: '5', path: '头像/原图.png',
|
|
size: 12, sha256: 'a'.repeat(64), key: 'migration/new.png',
|
|
url: 'https://oss/new.png', type: 'image/png', files: []
|
|
};
|
|
if (!orphan) for (let index = 0; index < count; index++) item.files.push({
|
|
id: String(9007199254741093n + BigInt(index)), config_id: '5', path: item.path,
|
|
url: 'https://old/old.png', type: 'image/png', size: 12
|
|
});
|
|
return [item];
|
|
}
|
|
|
|
class TransitionDb {
|
|
constructor(items) {
|
|
this.items = items;
|
|
this.calls = [];
|
|
this.files = Object.fromEntries(items.flatMap(item => item.files.map(file => [String(file.id), clone(file)])));
|
|
this.states = Object.fromEntries(items.map(item => [itemKey(item), 'verified']));
|
|
this.logs = [];
|
|
this.statusWarning = 0;
|
|
this.statusAffected = null;
|
|
this.fileAffected = null;
|
|
}
|
|
async beginTransaction() {
|
|
this.calls.push({ transaction: 'begin' });
|
|
if (this.beginFailure) throw this.beginFailure;
|
|
this.saved = clone({ files: this.files, states: this.states, logs: this.logs });
|
|
}
|
|
async commit() {
|
|
this.calls.push({ transaction: 'commit' });
|
|
if (this.commitFailure) throw this.commitFailure;
|
|
this.saved = undefined;
|
|
}
|
|
async rollback() {
|
|
this.calls.push({ transaction: 'rollback' });
|
|
if (this.rollbackFailure) throw this.rollbackFailure;
|
|
if (this.saved) Object.assign(this, this.saved);
|
|
this.saved = undefined;
|
|
}
|
|
async execute(sql, params = []) {
|
|
this.calls.push({ sql, params });
|
|
if (sql.startsWith('SELECT id FROM')) {
|
|
if (this.sourceFailure) throw this.sourceFailure;
|
|
return [this.items.map(item => ({ id: item.contentId }))];
|
|
}
|
|
if (sql.startsWith('SELECT id,CASE')) {
|
|
return [this.items.map(item => ({ id: this.newerSource ? String(BigInt(item.contentId) + 1n) : item.contentId, expected_content_id: item.contentId }))];
|
|
}
|
|
if (sql.startsWith('SELECT id,config_id,path,OCTET_LENGTH')) {
|
|
return [this.items.map(item => ({ id: item.contentId, config_id: item.configId, path: item.path, size: item.size, sha256: item.sha256 }))];
|
|
}
|
|
if (sql.includes('information_schema.COLUMNS')) {
|
|
return [[...['id', 'config_id', 'path', 'url', 'type', 'size', 'deleted', 'update_time'].map(name => ({
|
|
name, extra: name === 'update_time' ? 'on update CURRENT_TIMESTAMP' : '', engine: 'InnoDB'
|
|
}))]];
|
|
}
|
|
if (sql.startsWith('UPDATE') && sql.includes('`infra_file`')) {
|
|
const [set, where] = sql.split(' WHERE ');
|
|
let cursor = (set.match(/\?/g) || []).length;
|
|
const groups = where.split(' OR ');
|
|
for (const group of groups) {
|
|
const pk = String(params[cursor]);
|
|
cursor += (group.match(/\?/g) || []).length;
|
|
const item = this.items.find(value => value.files.some(file => String(file.id) === pk));
|
|
assert.ok(item, 'file updates must target the selected batch');
|
|
this.files[pk] = { ...this.files[pk], config_id: config.targetConfigId, path: item.key, url: item.url, type: item.type, size: item.size };
|
|
}
|
|
return [{ affectedRows: this.fileAffected ?? groups.length, warningStatus: 0 }];
|
|
}
|
|
if (sql.startsWith('SELECT `')) {
|
|
return [params.map(pk => this.files[String(pk)])];
|
|
}
|
|
if (sql.startsWith('INSERT INTO')) {
|
|
if (this.journalFailure) throw this.journalFailure;
|
|
for (let index = 0; index < params.length; index += 7) this.logs.push(params.slice(index, index + 7));
|
|
return [{ affectedRows: params.length / 7, warningStatus: 0 }];
|
|
}
|
|
if (sql.startsWith('UPDATE') && sql.includes('`infra_file_oss_migration_items`')) {
|
|
assert.match(sql, /WHERE run_id=\? AND status='verified' AND item_key IN \(/);
|
|
assert.match(sql, /SET status='switched',error=NULL,updated_at=CURRENT_TIMESTAMP/);
|
|
assert.equal(params[0], config.runId);
|
|
let affected = 0;
|
|
for (const key of params.slice(1)) if (this.states[key] === 'verified') { this.states[key] = 'switched'; affected++; }
|
|
return [{ affectedRows: this.statusAffected ?? affected, warningStatus: this.statusWarning }];
|
|
}
|
|
throw new Error(`Unexpected query: ${sql}`);
|
|
}
|
|
}
|
|
|
|
const state = db => clone({ files: db.files, states: db.states, logs: db.logs });
|
|
|
|
test('source range verification is the first operation after BEGIN, then files and guarded item status commit together', async () => {
|
|
const items = fixture();
|
|
const db = new TransitionDb(items);
|
|
const result = await switchFiles(db, config, items);
|
|
assert.deepEqual(Object.keys(result), ['sourceMs', 'writeMs', 'statusMs', 'commitMs']);
|
|
assert.ok(Object.values(result).every(value => Number.isFinite(value) && value >= 0));
|
|
assert.equal(db.calls[0].transaction, 'begin');
|
|
assert.match(db.calls[1].sql, /^SELECT id FROM .* WHERE id>=\? ORDER BY id FOR UPDATE$/);
|
|
assert.match(db.calls[2].sql, /^SELECT id,CASE/);
|
|
assert.match(db.calls[3].sql, /^SELECT id,config_id,path,OCTET_LENGTH/);
|
|
assert.ok(db.calls[4].sql.includes('information_schema.COLUMNS'));
|
|
assert.equal(db.calls.at(-1).transaction, 'commit');
|
|
assert.equal(db.files[items[0].files[0].id].url, items[0].url);
|
|
assert.equal(db.states[itemKey(items[0])], 'switched');
|
|
assert.equal(db.logs.length, 1);
|
|
const statusCall = db.calls.find(call => call.sql?.includes("SET status='switched'"));
|
|
assert.deepEqual(statusCall.params, [config.runId, itemKey(items[0])]);
|
|
});
|
|
|
|
test('more than 100 file records use bounded write chunks in one source-locked transaction', async () => {
|
|
const items = fixture({ count: 101 });
|
|
const db = new TransitionDb(items);
|
|
await switchFiles(db, config, items);
|
|
assert.deepEqual(db.calls.filter(call => call.transaction).map(call => call.transaction), ['begin', 'commit']);
|
|
assert.equal(db.calls.filter(call => call.sql?.startsWith('SELECT id FROM')).length, 1);
|
|
const writes = db.calls.filter(call => call.sql?.startsWith('UPDATE') && call.sql.includes('`infra_file`'));
|
|
assert.equal(writes.length, 2);
|
|
assert.equal(writes[0].sql.split(' WHERE ')[1].split(' OR ').length, 100);
|
|
assert.equal(writes[1].sql.split(' WHERE ')[1].split(' OR ').length, 1);
|
|
assert.equal(db.logs.length, 101);
|
|
assert.equal(Object.values(db.files).filter(file => file.url === items[0].url).length, 101);
|
|
});
|
|
|
|
test('orphan content is source-verified and status-switched even without any infra_file rows', async () => {
|
|
const items = fixture({ orphan: true });
|
|
const db = new TransitionDb(items);
|
|
await switchFiles(db, config, items);
|
|
assert.equal(db.states[itemKey(items[0])], 'switched');
|
|
assert.equal(db.logs.length, 0);
|
|
assert.equal(db.calls.filter(call => call.sql).length, 4);
|
|
assert.equal(db.calls.some(call => call.sql?.includes('information_schema')), false);
|
|
});
|
|
|
|
test('mixed regular and orphan items both participate in one guarded status update', async () => {
|
|
const items = fixture();
|
|
items.push({ ...fixture({ orphan: true })[0], contentId: '9007199254740994', path: 'orphan.png' });
|
|
const db = new TransitionDb(items);
|
|
await switchFiles(db, config, items);
|
|
assert.equal(db.logs.length, 1);
|
|
assert.deepEqual(Object.values(db.states), ['switched', 'switched']);
|
|
});
|
|
|
|
test('source errors are propagated unchanged after a successful rollback', async () => {
|
|
const items = fixture();
|
|
const db = new TransitionDb(items);
|
|
const failure = Object.assign(new Error('source changed'), { code: 'MIGRATION_SOURCE_CHANGED' });
|
|
db.sourceFailure = failure;
|
|
await assert.rejects(switchFiles(db, config, items), error => error === failure);
|
|
assert.deepEqual(db.calls.filter(call => call.transaction).map(call => call.transaction), ['begin', 'rollback']);
|
|
assert.equal(db.calls.some(call => call.sql?.startsWith('UPDATE')), false);
|
|
});
|
|
|
|
test('a newer source version stops the transaction before any file updates', async () => {
|
|
const items = fixture();
|
|
const db = new TransitionDb(items);
|
|
db.newerSource = true;
|
|
await assert.rejects(switchFiles(db, config, items), { code: 'MIGRATION_SOURCE_CHANGED' });
|
|
assert.equal(db.calls.some(call => call.sql?.startsWith('UPDATE')), false);
|
|
assert.equal(db.calls.at(-1).transaction, 'rollback');
|
|
});
|
|
|
|
test('file CAS mismatches roll back all file edits and preserve the conflict code', async () => {
|
|
const items = fixture();
|
|
const db = new TransitionDb(items);
|
|
const before = state(db);
|
|
db.fileAffected = 0;
|
|
await assert.rejects(switchFiles(db, config, items), { code: 'MIGRATION_CAS_CONFLICT' });
|
|
assert.deepEqual(state(db), before);
|
|
assert.equal(db.calls.some(call => call.sql?.includes("SET status='switched'")), false);
|
|
});
|
|
|
|
test('non-verified item status rolls back previously updated files and journal inserts', async () => {
|
|
const items = fixture();
|
|
const db = new TransitionDb(items);
|
|
db.states[itemKey(items[0])] = 'planned';
|
|
const before = state(db);
|
|
await assert.rejects(switchFiles(db, config, items), { code: 'MIGRATION_CAS_CONFLICT' });
|
|
assert.deepEqual(state(db), before);
|
|
assert.ok(db.calls.some(call => call.sql?.startsWith('INSERT INTO')));
|
|
});
|
|
|
|
test('item status SQL warnings roll back files, logs and item statuses', async () => {
|
|
const items = fixture();
|
|
const db = new TransitionDb(items);
|
|
const before = state(db);
|
|
db.statusWarning = 1;
|
|
await assert.rejects(switchFiles(db, config, items), { code: 'MIGRATION_SQL_WARNING' });
|
|
assert.deepEqual(state(db), before);
|
|
});
|
|
|
|
test('rollback failure is fatal and prohibits fallback while preserving both causes', async () => {
|
|
const items = fixture();
|
|
const db = new TransitionDb(items);
|
|
db.sourceFailure = Object.assign(new Error('source failed'), { code: 'MIGRATION_SOURCE_CHANGED' });
|
|
db.rollbackFailure = new Error('connection lost');
|
|
await assert.rejects(switchFiles(db, config, items), error => {
|
|
assert.equal(error.code, 'MIGRATION_ROLLBACK_FAILED');
|
|
assert.equal(error.fatal, true);
|
|
assert.equal(error.noFallback, true);
|
|
assert.equal(error.cause, db.sourceFailure);
|
|
assert.equal(error.rollbackError, db.rollbackFailure);
|
|
return true;
|
|
});
|
|
});
|
|
|
|
test('commit acknowledgement failure is fatal even if a rollback call subsequently succeeds', async () => {
|
|
const items = fixture();
|
|
const db = new TransitionDb(items);
|
|
db.commitFailure = Object.assign(new Error('commit transport failure'), { code: 'PROTOCOL_CONNECTION_LOST' });
|
|
await assert.rejects(switchFiles(db, config, items), error => {
|
|
assert.equal(error.code, 'PROTOCOL_CONNECTION_LOST');
|
|
assert.equal(error.fatal, true);
|
|
assert.equal(error.noFallback, true);
|
|
assert.equal(error.cause, db.commitFailure);
|
|
return true;
|
|
});
|
|
assert.deepEqual(db.calls.filter(call => call.transaction).map(call => call.transaction), ['begin', 'commit', 'rollback']);
|
|
});
|
|
|
|
test('BEGIN failure is propagated without performing any reads, writes or rollback', async () => {
|
|
const items = fixture();
|
|
const db = new TransitionDb(items);
|
|
db.beginFailure = new Error('cannot begin');
|
|
await assert.rejects(switchFiles(db, config, items), error => error === db.beginFailure);
|
|
assert.deepEqual(db.calls, [{ transaction: 'begin' }]);
|
|
});
|
|
|
|
test('empty batches are no-ops and malformed or duplicate input is rejected before BEGIN', async () => {
|
|
const db = new TransitionDb([]);
|
|
assert.deepEqual(await switchFiles(db, config, []), { sourceMs: 0, writeMs: 0, statusMs: 0, commitMs: 0 });
|
|
await assert.rejects(switchFiles(db, config, null), /数组/);
|
|
await assert.rejects(switchFiles(db, config, [{ ...fixture()[0], files: null }]), /files/);
|
|
await assert.rejects(switchFiles(db, config, [fixture()[0], fixture()[0]]), /重复源路径/);
|
|
const duplicateFiles = fixture();
|
|
duplicateFiles[0].files.push({ ...duplicateFiles[0].files[0] });
|
|
await assert.rejects(switchFiles(db, config, duplicateFiles), /重复文件记录/);
|
|
assert.equal(db.calls.length, 0);
|
|
});
|