312 lines
15 KiB
JavaScript
312 lines
15 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { applyChanges } from '../lib/journal.mjs';
|
|
|
|
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;
|
|
const key = value => Buffer.isBuffer(value) ? `buffer:${value.toString('hex')}` : String(value);
|
|
const same = (left, right) => Buffer.isBuffer(left) || Buffer.isBuffer(right)
|
|
? Buffer.isBuffer(left) && Buffer.isBuffer(right) && left.equals(right)
|
|
: left === null || right === null ? left === right : String(left) === String(right);
|
|
|
|
const makeRow = (id, suffix) => ({
|
|
id, config_id: '4', path: `${suffix}.png`, url: `https://old/${suffix}.png`, type: 'image/png', size: 5,
|
|
deleted: Buffer.from([0]), update_time: 'unchanged', touched: 'unchanged'
|
|
});
|
|
|
|
// This small SQL test double evaluates the CASE dispatch and every WHERE guard
|
|
// independently, so mixed per-row values and partially matched batches surface.
|
|
class BatchConnection {
|
|
constructor(rows = [makeRow('9007199254740993', 'a'), makeRow('9007199254740994', 'b')]) {
|
|
this.rows = Object.fromEntries(rows.map(row => [key(row.id), clone(row)]));
|
|
this.calls = [];
|
|
this.journals = [];
|
|
this.warningStatus = 0;
|
|
this.failJournal = false;
|
|
this.journalWarningStatus = 0;
|
|
this.journalAffectedRows = null;
|
|
this.engine = 'InnoDB';
|
|
this.normalize = null;
|
|
this.omitReadback = false;
|
|
this.reverseReadback = false;
|
|
this.nextId = 9007199254741993n;
|
|
}
|
|
|
|
async beginTransaction() {
|
|
this.calls.push({ transaction: 'begin' });
|
|
this.saved = { rows: clone(this.rows), journals: clone(this.journals) };
|
|
}
|
|
async commit() { this.calls.push({ transaction: 'commit' }); this.saved = undefined; }
|
|
async rollback() {
|
|
this.calls.push({ transaction: 'rollback' });
|
|
if (this.saved) { this.rows = this.saved.rows; this.journals = this.saved.journals; this.saved = undefined; }
|
|
}
|
|
|
|
async execute(sql, params = []) {
|
|
this.calls.push({ sql, params });
|
|
if (sql.includes('FROM information_schema.COLUMNS')) {
|
|
return [[...['id', 'config_id', 'path', 'url', 'type', 'size', 'deleted', 'json_data', 'counter', 'update_time', 'touched']
|
|
.map(name => ({ name, engine: this.engine, extra: ['update_time', 'touched'].includes(name) ? 'on update CURRENT_TIMESTAMP' : '' }))]];
|
|
}
|
|
if (sql.startsWith('UPDATE')) {
|
|
const [setSql, whereSql] = sql.split(/\bWHERE\b/);
|
|
const cases = [...setSql.matchAll(/`([^`]+)`\s*=\s*CASE\s+`id`([\s\S]*?)\bEND/g)];
|
|
assert.ok(cases.length > 0, 'the batch must use a CASE expression for changed columns');
|
|
let offset = 0;
|
|
const values = new Map();
|
|
for (const [, column, branches] of cases) {
|
|
const count = [...branches.matchAll(/\bWHEN\s+\?\s+THEN\s+\?/g)].length;
|
|
assert.ok(count > 0);
|
|
assert.match(branches, new RegExp('ELSE\\s+`' + column + '`'));
|
|
const entries = new Map();
|
|
for (let index = 0; index < count; index++) entries.set(key(params[offset++]), params[offset++]);
|
|
values.set(column, entries);
|
|
}
|
|
let affectedRows = 0;
|
|
const groups = whereSql.trim().replace(/^\(/, '').replace(/\)$/, '').split(/\)\s+OR\s+\(/);
|
|
for (const group of groups) {
|
|
assert.match(group, /`id`\s*=\s*\?/);
|
|
const pk = key(params[offset++]);
|
|
const guards = [...group.matchAll(/CAST\(`([^`]+)` AS BINARY\)\s*<=>\s*CAST\(\? AS BINARY\)/g)].map(match => match[1]);
|
|
const expected = guards.map(() => params[offset++]);
|
|
const row = this.rows[pk];
|
|
if (!row || !guards.every((name, index) => same(row[name], expected[index]))) continue;
|
|
for (const [column, entries] of values) if (entries.has(pk)) row[column] = entries.get(pk);
|
|
if (this.normalize) this.normalize(row, [...values.keys()]);
|
|
if (!setSql.includes('`update_time` = `update_time`')) row.update_time = 'changed';
|
|
if (!setSql.includes('`touched` = `touched`')) row.touched = 'changed';
|
|
affectedRows++;
|
|
}
|
|
assert.equal(offset, params.length, 'all prepared UPDATE values must have a matching SQL placeholder');
|
|
return [{ affectedRows, warningStatus: this.warningStatus }];
|
|
}
|
|
if (sql.startsWith('SELECT `')) {
|
|
assert.match(sql, /\bIN\s*\(/);
|
|
assert.match(sql, /FOR UPDATE/);
|
|
const selected = [...sql.split(/\bFROM\b/)[0].matchAll(/`([^`]+)`/g)].map(match => match[1]);
|
|
const rows = params.map(pk => this.rows[key(pk)]).filter(Boolean)
|
|
.map(row => Object.fromEntries(selected.map(column => [column, row[column]])));
|
|
if (this.reverseReadback) rows.reverse();
|
|
return [this.omitReadback ? rows.slice(1) : rows];
|
|
}
|
|
if (sql.startsWith('INSERT INTO')) {
|
|
if (this.failJournal) throw Object.assign(new Error('injected journal INSERT failure'), { code: 'TEST_JOURNAL_FAILURE' });
|
|
assert.equal(params.length % 7, 0);
|
|
const firstId = this.nextId.toString();
|
|
for (let offset = 0; offset < params.length; offset += 7) {
|
|
const [run_id, schema_name, table_name, pk_name, pk_value, before_json, after_json] = params.slice(offset, offset + 7);
|
|
this.journals.push({ id: (this.nextId++).toString(), run_id, schema_name, table_name, pk_name, pk_value, before_json, after_json, rolled_back: 0 });
|
|
}
|
|
return [{ affectedRows: this.journalAffectedRows ?? params.length / 7, warningStatus: this.journalWarningStatus, insertId: firstId }];
|
|
}
|
|
throw new Error(`Unexpected test SQL: ${sql}`);
|
|
}
|
|
}
|
|
|
|
function change(row, suffix = 'migrated') {
|
|
const { config_id, path, url, type, size, deleted } = row;
|
|
return {
|
|
schema: 'app', table: 'infra_file', primaryKey: 'id', pkValue: row.id,
|
|
before: { config_id, path, url, type, size, deleted },
|
|
after: { config_id: '20', path: `${suffix}/${path}`, url: `https://oss/${suffix}/${path}`, type, size }
|
|
};
|
|
}
|
|
|
|
const changesFor = connection => Object.values(connection.rows).map(row => change(row));
|
|
const dml = connection => connection.calls.filter(call => call.sql && !call.sql.includes('information_schema'));
|
|
|
|
async function commitBatch(connection, changes) {
|
|
await connection.beginTransaction();
|
|
try {
|
|
const result = await applyChanges(connection, 'app', 'run', changes);
|
|
await connection.commit();
|
|
return result;
|
|
} catch (error) { await connection.rollback(); throw error; }
|
|
}
|
|
|
|
test('batch uses one CASE update, one readback and one journal insert for distinct bigint rows', async () => {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
assert.deepEqual(await commitBatch(connection, changes), { changed: 2, skipped: 0 });
|
|
assert.deepEqual(dml(connection).map(call => call.sql.split(/\s/)[0]), ['UPDATE', 'SELECT', 'INSERT']);
|
|
assert.equal(connection.rows['9007199254740993'].url, 'https://oss/migrated/a.png');
|
|
assert.equal(connection.rows['9007199254740994'].url, 'https://oss/migrated/b.png');
|
|
assert.equal(connection.rows['9007199254740993'].update_time, 'unchanged');
|
|
assert.equal(connection.rows['9007199254740994'].touched, 'unchanged');
|
|
assert.equal(connection.journals.length, 2);
|
|
assert.deepEqual(connection.journals.map(row => JSON.parse(row.pk_value).value), ['9007199254740993', '9007199254740994']);
|
|
const [update] = dml(connection);
|
|
assert.match(update.sql, /CAST\(`deleted` AS BINARY\) <=> CAST\(\? AS BINARY\)/);
|
|
assert.ok(update.params.some(value => Buffer.isBuffer(value) && value.equals(Buffer.from([0]))));
|
|
assert.ok(!update.sql.includes('https://oss/'), 'URLs must be bound, not interpolated');
|
|
assert.ok(!update.sql.includes('`deleted` = CASE'), 'before-only deletion guards must not become assignments');
|
|
});
|
|
|
|
test('table metadata is cached across batches on the same connection', async () => {
|
|
const connection = new BatchConnection();
|
|
await commitBatch(connection, changesFor(connection));
|
|
await commitBatch(connection, Object.values(connection.rows).map(row => change(row, 'second')));
|
|
assert.equal(connection.calls.filter(call => call.sql?.includes('information_schema')).length, 1);
|
|
assert.equal(dml(connection).length, 6);
|
|
});
|
|
|
|
test('empty batches and unchanged rows create no updates or journal entries', async () => {
|
|
const connection = new BatchConnection();
|
|
assert.deepEqual(await commitBatch(connection, []), { changed: 0, skipped: 0 });
|
|
const changes = changesFor(connection);
|
|
changes.forEach(item => { item.after = Object.fromEntries(Object.keys(item.after).map(name => [name, item.before[name]])); });
|
|
assert.deepEqual(await commitBatch(connection, changes), { changed: 0, skipped: 2 });
|
|
assert.equal(dml(connection).length, 0);
|
|
assert.equal(connection.journals.length, 0);
|
|
});
|
|
|
|
test('a no-op row is skipped while a different row receives its own values', async () => {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
changes[1].after = Object.fromEntries(Object.keys(changes[1].after).map(name => [name, changes[1].before[name]]));
|
|
assert.deepEqual(await commitBatch(connection, changes), { changed: 1, skipped: 1 });
|
|
assert.equal(connection.rows['9007199254740993'].url, 'https://oss/migrated/a.png');
|
|
assert.equal(connection.rows['9007199254740994'].url, 'https://old/b.png');
|
|
assert.equal(connection.journals.length, 1);
|
|
});
|
|
|
|
test('after field order may differ and row-specific unchanged fields stay unchanged', async () => {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
changes[0].after.type = 'image/webp';
|
|
changes[1].after = Object.fromEntries(Object.entries(changes[1].after).reverse());
|
|
await commitBatch(connection, changes);
|
|
assert.equal(connection.rows['9007199254740993'].type, 'image/webp');
|
|
assert.equal(connection.rows['9007199254740994'].type, 'image/png');
|
|
});
|
|
|
|
test('different schema, table, primary key, or after-field groups are rejected before updates', async () => {
|
|
for (const mutate of [
|
|
item => { item.schema = 'other'; },
|
|
item => { item.table = 'member_user'; },
|
|
item => { item.primaryKey = 'other_id'; },
|
|
item => { delete item.after.size; }
|
|
]) {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
mutate(changes[1]);
|
|
await assert.rejects(commitBatch(connection, changes));
|
|
assert.equal(dml(connection).length, 0);
|
|
}
|
|
});
|
|
|
|
test('duplicate primary keys are rejected even across string and bigint representations', async () => {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
changes[1].pkValue = BigInt(changes[0].pkValue);
|
|
await assert.rejects(commitBatch(connection, changes), /主键|重复/);
|
|
assert.equal(dml(connection).length, 0);
|
|
});
|
|
|
|
test('primary key updates and missing before snapshots are rejected', async () => {
|
|
for (const mutate of [
|
|
item => { item.after.id = '1'; },
|
|
item => { delete item.before.url; }
|
|
]) {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
changes.forEach(mutate);
|
|
await assert.rejects(commitBatch(connection, changes));
|
|
assert.equal(dml(connection).length, 0);
|
|
}
|
|
});
|
|
|
|
test('a concurrent case-only edit leaves the entire batch unchanged after caller rollback', async () => {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
connection.rows['9007199254740994'].url = 'https://old/B.png';
|
|
const before = clone(connection.rows);
|
|
await assert.rejects(commitBatch(connection, changes), { code: 'MIGRATION_CAS_CONFLICT' });
|
|
assert.deepEqual(connection.rows, before);
|
|
assert.equal(connection.journals.length, 0);
|
|
assert.equal(dml(connection).length, 1);
|
|
assert.equal(connection.calls.at(-1).transaction, 'rollback');
|
|
});
|
|
|
|
test('a deletion guard conflict rolls back otherwise matching rows', async () => {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
connection.rows['9007199254740994'].deleted = Buffer.from([1]);
|
|
const before = clone(connection.rows);
|
|
await assert.rejects(commitBatch(connection, changes), { code: 'MIGRATION_CAS_CONFLICT' });
|
|
assert.deepEqual(connection.rows, before);
|
|
});
|
|
|
|
test('journal INSERT failure rolls back all successfully updated rows', async () => {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
const before = clone(connection.rows);
|
|
connection.failJournal = true;
|
|
await assert.rejects(commitBatch(connection, changes), { code: 'TEST_JOURNAL_FAILURE' });
|
|
assert.deepEqual(connection.rows, before);
|
|
assert.equal(connection.journals.length, 0);
|
|
assert.equal(dml(connection).length, 3);
|
|
});
|
|
|
|
test('MySQL JSON canonicalization and numeric conversion are read back into each journal entry', async () => {
|
|
const connection = new BatchConnection();
|
|
connection.reverseReadback = true;
|
|
for (const row of Object.values(connection.rows)) {
|
|
row.json_data = JSON.stringify({ url: row.url }).replaceAll('\":', '\": ');
|
|
row.counter = 1;
|
|
}
|
|
const changes = Object.values(connection.rows).map(row => ({
|
|
schema: 'app', table: 'infra_file', primaryKey: 'id', pkValue: row.id,
|
|
before: { json_data: row.json_data, counter: 1 },
|
|
after: { json_data: JSON.stringify({ url: `https://oss/${row.path}` }), counter: '2' }
|
|
}));
|
|
connection.normalize = (row, fields) => {
|
|
if (fields.includes('json_data')) row.json_data = JSON.stringify(JSON.parse(row.json_data)).replaceAll('\":', '\": ');
|
|
if (fields.includes('counter')) row.counter = Number(row.counter);
|
|
};
|
|
await commitBatch(connection, changes);
|
|
const snapshots = connection.journals.map(row => JSON.parse(row.after_json));
|
|
assert.equal(snapshots[0].json_data.value, '{"url": "https://oss/a.png"}');
|
|
assert.equal(snapshots[1].json_data.value, '{"url": "https://oss/b.png"}');
|
|
assert.equal(snapshots[0].counter.value, 2);
|
|
assert.equal(snapshots[1].counter.value, 2);
|
|
});
|
|
|
|
test('UPDATE warnings and incomplete readback each abort the entire batch', async () => {
|
|
for (const prepare of [
|
|
connection => { connection.warningStatus = 1; },
|
|
connection => { connection.omitReadback = true; }
|
|
]) {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
const before = clone(connection.rows);
|
|
prepare(connection);
|
|
await assert.rejects(commitBatch(connection, changes));
|
|
assert.deepEqual(connection.rows, before);
|
|
assert.equal(connection.journals.length, 0);
|
|
}
|
|
});
|
|
|
|
test('incomplete or warned batch journal inserts roll back both updated rows and inserted logs', async () => {
|
|
for (const prepare of [
|
|
connection => { connection.journalAffectedRows = 1; },
|
|
connection => { connection.journalWarningStatus = 1; }
|
|
]) {
|
|
const connection = new BatchConnection();
|
|
const changes = changesFor(connection);
|
|
const before = clone(connection.rows);
|
|
prepare(connection);
|
|
await assert.rejects(commitBatch(connection, changes), /回滚日志/);
|
|
assert.deepEqual(connection.rows, before);
|
|
assert.equal(connection.journals.length, 0);
|
|
}
|
|
});
|
|
|
|
test('non-InnoDB targets are rejected before mutation', async () => {
|
|
const connection = new BatchConnection();
|
|
connection.engine = 'MyISAM';
|
|
await assert.rejects(commitBatch(connection, changesFor(connection)), /InnoDB/);
|
|
assert.equal(dml(connection).length, 0);
|
|
});
|