214 lines
13 KiB
JavaScript
214 lines
13 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { setTimeout as delay } from 'node:timers/promises';
|
|
import { encodePath, sha256 } from '../lib/core.mjs';
|
|
import { ensureJournal, rollbackChanges } from '../lib/journal.mjs';
|
|
import { verifySourceBatch } from '../lib/source-batch.mjs';
|
|
import { switchFiles } from '../lib/transition.mjs';
|
|
|
|
// Opt in only against the disposable localhost test server. No host, database,
|
|
// credentials or application configuration is read from the project.
|
|
const SCHEMA = 'codex_oss_transition_test';
|
|
const PORT = process.env.MIGRATION_TEST_DB_PORT;
|
|
const IDS = {
|
|
older: '9007199254741001', first: '9007199254741003', second: '9007199254741005',
|
|
orphan: '9007199254741007', deletedNewer: '9007199254741009', newVersion: '9007199254741011'
|
|
};
|
|
const FIRST_PATH = '相册/Résumé.PNG';
|
|
const EQUIVALENT_PATH = '相册/RESUME.png';
|
|
const imageBytes = [Buffer.from([1, 2, 3, 4, 5]), Buffer.from([6, 7, 8, 9]), Buffer.from([10, 11, 12])];
|
|
const changedSource = error => error.code === 'MIGRATION_SOURCE_CHANGED';
|
|
const changedRow = error => error.code === 'MIGRATION_CAS_CONFLICT';
|
|
|
|
test('MySQL batch transitions preserve source locks, atomic journals and rollback', {
|
|
skip: !PORT, timeout: 45000
|
|
}, async t => {
|
|
const port = Number(PORT);
|
|
assert.ok(Number.isInteger(port) && port > 0 && port <= 65535, 'MIGRATION_TEST_DB_PORT must be a valid disposable localhost MySQL port');
|
|
const { default: mysql } = await import('mysql2/promise');
|
|
const connectionOptions = {
|
|
host: '127.0.0.1', port, user: 'root', password: '', connectTimeout: 3000,
|
|
supportBigNumbers: true, bigNumberStrings: true, jsonStrings: true, dateStrings: true
|
|
};
|
|
const db = await mysql.createConnection(connectionOptions);
|
|
let contender, observer;
|
|
try {
|
|
await db.query(`DROP DATABASE IF EXISTS ${SCHEMA}`);
|
|
await db.query(`CREATE DATABASE ${SCHEMA} CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`);
|
|
await db.changeUser({ database: SCHEMA });
|
|
await db.query('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
|
|
await db.query("SET SESSION sql_mode=CONCAT_WS(',',NULLIF(@@SESSION.sql_mode,''),'STRICT_ALL_TABLES')");
|
|
await db.query('SET SESSION innodb_lock_wait_timeout=5');
|
|
await db.query(`CREATE TABLE infra_file_content (
|
|
id BIGINT UNSIGNED NOT NULL PRIMARY KEY,config_id BIGINT UNSIGNED NOT NULL,
|
|
path VARCHAR(512) NOT NULL,content MEDIUMBLOB NOT NULL,deleted BIT(1) NOT NULL DEFAULT 0
|
|
) ENGINE=InnoDB`);
|
|
await db.query(`CREATE TABLE infra_file (
|
|
id BIGINT UNSIGNED NOT NULL PRIMARY KEY,config_id BIGINT UNSIGNED NOT NULL,
|
|
path VARCHAR(512) NOT NULL,url VARCHAR(1024) NOT NULL,type VARCHAR(128),size INT,
|
|
deleted BIT(1) NOT NULL DEFAULT 0,
|
|
update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
|
|
) ENGINE=InnoDB`);
|
|
await ensureJournal(db, SCHEMA);
|
|
|
|
contender = await mysql.createConnection({ ...connectionOptions, database: SCHEMA });
|
|
observer = await mysql.createConnection({ ...connectionOptions, database: SCHEMA });
|
|
await contender.query('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
|
|
await contender.query('SET SESSION innodb_lock_wait_timeout=5');
|
|
const [[thread]] = await contender.query('SELECT CONNECTION_ID() AS id');
|
|
|
|
const readFiles = async () => {
|
|
const [rows] = await db.query('SELECT id,config_id,path,url,type,size,deleted,update_time FROM infra_file ORDER BY id');
|
|
return rows.map(row => ({ ...row }));
|
|
};
|
|
const readStatuses = async runId => {
|
|
const [rows] = await db.execute('SELECT status FROM infra_file_oss_migration_items WHERE run_id=? ORDER BY item_key', [runId]);
|
|
return rows.map(row => row.status);
|
|
};
|
|
const journalCount = async runId => {
|
|
const [[row]] = await db.execute('SELECT COUNT(*) AS n FROM infra_file_oss_migration_changes WHERE run_id=?', [runId]);
|
|
return Number(row.n);
|
|
};
|
|
|
|
async function seed(runId) {
|
|
await db.rollback();
|
|
for (const table of ['infra_file_oss_migration_changes', 'infra_file_oss_migration_items', 'infra_file_oss_migration_runs', 'infra_file', 'infra_file_content']) {
|
|
await db.query(`DELETE FROM ${table}`);
|
|
}
|
|
const sources = [
|
|
[IDS.older, EQUIVALENT_PATH, Buffer.from('older version'), 0],
|
|
[IDS.first, FIRST_PATH, imageBytes[0], 0],
|
|
[IDS.second, 'second.png', imageBytes[1], 0],
|
|
[IDS.orphan, 'orphan.png', imageBytes[2], 0],
|
|
[IDS.deletedNewer, EQUIVALENT_PATH, Buffer.from('deleted newer version'), 1]
|
|
];
|
|
await db.execute(`INSERT INTO infra_file_content(id,config_id,path,content,deleted) VALUES ${sources.map(() => '(?,5,?,?,?)').join(',')}`, sources.flat());
|
|
const files = [
|
|
['9007199254741101', FIRST_PATH], ['9007199254741102', EQUIVALENT_PATH], ['9007199254741103', 'second.png']
|
|
];
|
|
await db.execute(`INSERT INTO infra_file(id,config_id,path,url,type,size,update_time) VALUES ${files.map(() => "(?,5,?,?,'image/png',1,'2020-01-01 00:00:00')").join(',')}`,
|
|
files.flatMap(([fileId, path]) => [fileId, path, `https://legacy.test/admin-api/infra/file/5/get/${encodePath(path)}`]));
|
|
const items = [];
|
|
for (const [index, [contentId, path]] of [[IDS.first, FIRST_PATH], [IDS.second, 'second.png'], [IDS.orphan, 'orphan.png']].entries()) {
|
|
const [records] = await db.execute('SELECT id,config_id,path,url,type,size FROM infra_file WHERE config_id=5 AND path=? AND deleted=0 ORDER BY id', [path]);
|
|
const key = `migration/test/${contentId}.png`;
|
|
items.push({ contentId, configId: '5', path, size: imageBytes[index].length, sha256: sha256(imageBytes[index]), key, url: `https://oss.test/${key}`, type: 'image/png', files: records.map(row => ({ ...row })) });
|
|
}
|
|
await db.execute("INSERT INTO infra_file_oss_migration_runs(run_id,settings,status) VALUES (?,?,'active')", [runId, JSON.stringify({ test: true })]);
|
|
for (const item of items) {
|
|
await db.execute("INSERT INTO infra_file_oss_migration_items(run_id,item_key,data,status) VALUES (?,?,?,'verified')", [runId, sha256(`${item.configId}\0${item.path}`), JSON.stringify(item)]);
|
|
}
|
|
return { config: { mysql: { database: SCHEMA }, runId, targetConfigId: '30' }, items, originals: await readFiles() };
|
|
}
|
|
|
|
await t.test('batch success switches duplicate metadata and orphan progress; journals restore original values', async () => {
|
|
const fixture = await seed('batch-success');
|
|
assert.equal(fixture.items[0].files.length, 2, 'native collation includes the case/accent-equivalent metadata');
|
|
assert.equal(fixture.items[2].files.length, 0, 'orphan content remains a migration item');
|
|
await switchFiles(db, fixture.config, fixture.items);
|
|
const files = await readFiles();
|
|
for (const item of fixture.items) for (const original of item.files) {
|
|
const stored = files.find(file => file.id === original.id);
|
|
assert.equal(stored.config_id, '30');
|
|
assert.equal(stored.path, item.key);
|
|
assert.equal(stored.url, item.url);
|
|
assert.equal(stored.size, item.size);
|
|
assert.equal(stored.update_time, '2020-01-01 00:00:00');
|
|
}
|
|
assert.deepEqual(await readStatuses(fixture.config.runId), ['switched', 'switched', 'switched']);
|
|
assert.equal(await journalCount(fixture.config.runId), 3);
|
|
const result = await rollbackChanges(db, SCHEMA, fixture.config.runId);
|
|
assert.equal(result.restored, 3);
|
|
assert.deepEqual(result.conflicts, []);
|
|
assert.deepEqual(await readFiles(), fixture.originals);
|
|
const [[source]] = await db.execute('SELECT content FROM infra_file_content WHERE id=?', [IDS.first]);
|
|
assert.deepEqual(source.content, imageBytes[0]);
|
|
});
|
|
|
|
await t.test('a business-row CAS conflict rolls back the entire batch and preserves concurrent edits', async () => {
|
|
const fixture = await seed('cas-conflict');
|
|
await db.execute('UPDATE infra_file SET url=? WHERE id=?', ['https://user-edit.test/new.png', fixture.items[1].files[0].id]);
|
|
const beforeAttempt = await readFiles();
|
|
await assert.rejects(switchFiles(db, fixture.config, fixture.items), changedRow);
|
|
assert.deepEqual(await readFiles(), beforeAttempt);
|
|
assert.deepEqual(await readStatuses(fixture.config.runId), ['verified', 'verified', 'verified']);
|
|
assert.equal(await journalCount(fixture.config.runId), 0);
|
|
});
|
|
|
|
await t.test('a final progress-state CAS conflict rolls back already-written files and journals', async () => {
|
|
const fixture = await seed('status-conflict');
|
|
const orphan = fixture.items[2];
|
|
await db.execute("UPDATE infra_file_oss_migration_items SET status='failed' WHERE run_id=? AND item_key=?", [fixture.config.runId, sha256(`${orphan.configId}\0${orphan.path}`)]);
|
|
await assert.rejects(switchFiles(db, fixture.config, fixture.items), changedRow);
|
|
assert.deepEqual(await readFiles(), fixture.originals);
|
|
assert.deepEqual((await readStatuses(fixture.config.runId)).sort(), ['failed', 'verified', 'verified']);
|
|
assert.equal(await journalCount(fixture.config.runId), 0);
|
|
});
|
|
|
|
await t.test('source mutations and a newer collation-equivalent version prevent every database switch', async t2 => {
|
|
for (const kind of ['same-size-hash-change', 'deleted-source', 'newer-equivalent-path']) {
|
|
await t2.test(kind, async () => {
|
|
const fixture = await seed(kind);
|
|
if (kind === 'same-size-hash-change') await db.execute('UPDATE infra_file_content SET content=? WHERE id=?', [Buffer.from([9, 9, 9, 9, 9]), IDS.first]);
|
|
else if (kind === 'deleted-source') await db.execute('UPDATE infra_file_content SET deleted=1 WHERE id=?', [IDS.first]);
|
|
else await db.execute('INSERT INTO infra_file_content(id,config_id,path,content) VALUES (?,5,?,?)', [IDS.newVersion, EQUIVALENT_PATH, Buffer.from('new active version')]);
|
|
await assert.rejects(switchFiles(db, fixture.config, fixture.items), changedSource);
|
|
assert.deepEqual(await readFiles(), fixture.originals);
|
|
assert.deepEqual(await readStatuses(fixture.config.runId), ['verified', 'verified', 'verified']);
|
|
assert.equal(await journalCount(fixture.config.runId), 0);
|
|
});
|
|
}
|
|
});
|
|
|
|
async function proveRangeLock(kind, statement, parameters) {
|
|
const fixture = await seed(`range-${kind}`);
|
|
let outcome, pending;
|
|
await db.beginTransaction();
|
|
try {
|
|
// Must be the first consistent read after BEGIN; the locking read
|
|
// inside verification establishes protection before the RR snapshot.
|
|
await verifySourceBatch(db, fixture.config, [fixture.items[0]]);
|
|
pending = contender.execute(statement, parameters).then(
|
|
value => { outcome = { value }; }, error => { outcome = { error }; }
|
|
);
|
|
const deadline = Date.now() + 3000;
|
|
let observedWait = false;
|
|
while (Date.now() < deadline) {
|
|
assert.equal(outcome, undefined, 'concurrent source mutation must not complete before commit');
|
|
const [transactions] = await observer.execute('SELECT trx_state FROM information_schema.innodb_trx WHERE trx_mysql_thread_id=?', [thread.id]);
|
|
if (transactions.some(transaction => transaction.trx_state === 'LOCK WAIT')) { observedWait = true; break; }
|
|
await delay(20);
|
|
}
|
|
assert.ok(observedWait, 'observe an actual InnoDB lock wait, not merely a delayed promise');
|
|
assert.equal(outcome, undefined);
|
|
await db.commit();
|
|
await pending;
|
|
if (outcome.error) throw outcome.error;
|
|
assert.equal(outcome.value[0].affectedRows, 1, 'mutation completes after the source lock is released');
|
|
} finally {
|
|
await db.rollback();
|
|
if (pending) await pending;
|
|
}
|
|
// Once the waiting writer commits, the next batch must detect the new
|
|
// active version rather than continuing with its old snapshot.
|
|
await db.beginTransaction();
|
|
try {
|
|
await assert.rejects(verifySourceBatch(db, fixture.config, [fixture.items[0]]), changedSource);
|
|
} finally { await db.rollback(); }
|
|
}
|
|
|
|
await t.test('RR range locks block insertion of a new matching version until commit', async () => {
|
|
await proveRangeLock('insert', 'INSERT INTO infra_file_content(id,config_id,path,content) VALUES (?,5,?,?)', [IDS.newVersion, EQUIVALENT_PATH, Buffer.from('concurrent version')]);
|
|
});
|
|
await t.test('RR range locks include deleted rows and block undeleting a newer version until commit', async () => {
|
|
await proveRangeLock('undelete', 'UPDATE infra_file_content SET deleted=0 WHERE id=?', [IDS.deletedNewer]);
|
|
});
|
|
} finally {
|
|
await db.rollback().catch(() => {});
|
|
if (contender) await contender.end();
|
|
if (observer) await observer.end();
|
|
try { await db.query(`DROP DATABASE IF EXISTS ${SCHEMA}`); }
|
|
finally { await db.end(); }
|
|
}
|
|
});
|