246 lines
11 KiB
JavaScript
246 lines
11 KiB
JavaScript
import test, { mock } from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { createServer } from 'node:http';
|
|
import { once } from 'node:events';
|
|
import { setImmediate } from 'node:timers/promises';
|
|
import { sha256 } from '../lib/core.mjs';
|
|
import { verifyCheckpoint, verifySwitchedBatch } from '../migrate.mjs';
|
|
|
|
const CONTENT = Buffer.from('89504e470d0a1a0a0011223344556677', 'hex');
|
|
const CORRUPT = Buffer.alloc(CONTENT.length, 1);
|
|
const CONFIG = { mysql: { database: 'target_test' }, targetConfigId: '20', timeoutMs: 2000, maxFileBytes: 1024, retries: 1 };
|
|
const item = (url, files = [{ id: '9007199254740993' }]) => ({
|
|
contentId: '9007199254740995', configId: '1', path: 'source.png',
|
|
key: 'migration/object.png', url, size: CONTENT.length, sha256: sha256(CONTENT), files,
|
|
});
|
|
const stored = entry => ({
|
|
id: entry.files[0].id, config_id: '20', path: entry.key,
|
|
url: entry.url, deleted: Buffer.from([0]),
|
|
});
|
|
const oss = (bytes = CONTENT) => ({
|
|
head: mock.fn(async () => ({ res: { headers: { 'content-length': String(bytes.length) } } })),
|
|
get: mock.fn(async () => ({ content: bytes })),
|
|
put: mock.fn(async () => { throw new Error('validation must never upload'); }),
|
|
});
|
|
|
|
async function server(t, handler) {
|
|
const instance = createServer(handler);
|
|
instance.listen(0, '127.0.0.1');
|
|
await once(instance, 'listening');
|
|
t.after(async () => {
|
|
const closing = new Promise((resolve, reject) => instance.close(error => error ? reject(error) : resolve()));
|
|
instance.closeAllConnections();
|
|
await closing;
|
|
});
|
|
return 'http://127.0.0.1:' + instance.address().port;
|
|
}
|
|
|
|
test('verified checkpoint validates authenticated and public bytes without PUT or source access', async t => {
|
|
let publicReads = 0;
|
|
const url = await server(t, (_request, response) => { publicReads++; response.end(CONTENT); });
|
|
const entry = item(url);
|
|
const client = oss();
|
|
await verifyCheckpoint(client, CONFIG, entry);
|
|
assert.equal(client.head.mock.callCount(), 1);
|
|
assert.equal(client.get.mock.callCount(), 1);
|
|
assert.equal(client.put.mock.callCount(), 0);
|
|
assert.deepEqual(client.head.mock.calls[0].arguments, [entry.key]);
|
|
assert.deepEqual(client.get.mock.calls[0].arguments, [entry.key]);
|
|
assert.equal(publicReads, 1);
|
|
});
|
|
|
|
test('missing checkpoint object is rejected without downloading or creating another object', async () => {
|
|
const client = oss();
|
|
const missing = Object.assign(new Error('missing object'), { code: 'NoSuchKey', status: 404 });
|
|
client.head = mock.fn(async () => { throw missing; });
|
|
await assert.rejects(verifyCheckpoint(client, CONFIG, item('http://127.0.0.1:1/unused')), error => error === missing);
|
|
assert.equal(client.get.mock.callCount(), 0);
|
|
assert.equal(client.put.mock.callCount(), 0);
|
|
});
|
|
|
|
test('checkpoint HEAD rejects mismatched sizes or configured size limits before GET', async () => {
|
|
for (const tooLarge of [false, true]) {
|
|
const client = oss();
|
|
if (!tooLarge) client.head = mock.fn(async () => ({ res: { headers: { 'content-length': String(CONTENT.length + 1) } } }));
|
|
await assert.rejects(verifyCheckpoint(client, { ...CONFIG, maxFileBytes: tooLarge ? 1 : CONFIG.maxFileBytes }, item('http://127.0.0.1:1/unused')), /长度/);
|
|
assert.equal(client.get.mock.callCount(), 0);
|
|
assert.equal(client.put.mock.callCount(), 0);
|
|
}
|
|
});
|
|
|
|
test('same-length authenticated corruption rejects before public validation', async t => {
|
|
let publicReads = 0;
|
|
const url = await server(t, (_request, response) => { publicReads++; response.end(CONTENT); });
|
|
const client = oss(CORRUPT);
|
|
await assert.rejects(verifyCheckpoint(client, CONFIG, item(url)), /SHA-256/);
|
|
assert.equal(publicReads, 0);
|
|
assert.equal(client.put.mock.callCount(), 0);
|
|
});
|
|
|
|
test('public corruption or access denial prevents reusing an otherwise valid checkpoint', async t => {
|
|
const url = await server(t, (request, response) => {
|
|
if (request.url === '/denied') { response.writeHead(403); response.end(); }
|
|
else response.end(CORRUPT);
|
|
});
|
|
for (const [path, message] of [['/corrupt', /SHA-256/], ['/denied', /HTTP 403/]]) {
|
|
const client = oss();
|
|
await assert.rejects(verifyCheckpoint(client, CONFIG, item(url + path)), message);
|
|
assert.equal(client.put.mock.callCount(), 0);
|
|
}
|
|
});
|
|
|
|
test('switched batch reads all metadata once with exact bigint IDs and deduplicates repeated records', async t => {
|
|
const url = await server(t, (_request, response) => response.end(CONTENT));
|
|
const first = item(url + '/one');
|
|
const second = { ...item(url + '/two', [{ id: '9007199254740997' }, { id: '9007199254740997' }]), key: 'migration/two.png' };
|
|
const db = { execute: mock.fn(async (sql, params) => {
|
|
assert.match(sql, /SELECT id,config_id,path,url,deleted/);
|
|
assert.match(sql, /WHERE id IN \(\?,\?\)/);
|
|
assert.deepEqual(params, ['9007199254740993', '9007199254740997']);
|
|
return [[stored(first), stored(second)]];
|
|
}) };
|
|
const results = await verifySwitchedBatch(db, CONFIG, [first, second]);
|
|
assert.equal(db.execute.mock.callCount(), 1);
|
|
assert.deepEqual(results, [{ item: first }, { item: second }]);
|
|
});
|
|
|
|
test('orphaned source content with no file metadata still validates its public target', async t => {
|
|
let publicReads = 0;
|
|
const url = await server(t, (_request, response) => { publicReads++; response.end(CONTENT); });
|
|
const entry = item(url, []);
|
|
const db = { execute: mock.fn(async () => { throw new Error('orphan must not query file metadata'); }) };
|
|
assert.deepEqual(await verifySwitchedBatch(db, CONFIG, [entry]), [{ item: entry }]);
|
|
assert.equal(db.execute.mock.callCount(), 0);
|
|
assert.equal(publicReads, 1);
|
|
});
|
|
|
|
test('mutated, deleted and missing file records are returned as errors before public access', async t => {
|
|
let publicReads = 0;
|
|
const url = await server(t, (_request, response) => { publicReads++; response.end(CONTENT); });
|
|
const entry = item(url);
|
|
for (const changed of [
|
|
null,
|
|
{ ...stored(entry), config_id: '21' },
|
|
{ ...stored(entry), path: 'some/other/object.png' },
|
|
{ ...stored(entry), url: url + '/edited' },
|
|
{ ...stored(entry), deleted: Buffer.from([1]) },
|
|
]) {
|
|
const db = { execute: mock.fn(async () => [changed ? [changed] : []]) };
|
|
const results = await verifySwitchedBatch(db, CONFIG, [entry]);
|
|
assert.equal(results.length, 1);
|
|
assert.equal(results[0].item, entry);
|
|
assert.match(results[0].error.message, /发生变化|已删除/);
|
|
assert.deepEqual(results.filter(result => !result.error), []);
|
|
}
|
|
assert.equal(publicReads, 0);
|
|
});
|
|
|
|
test('a single missing metadata record rejects an item that has multiple original records', async t => {
|
|
let publicReads = 0;
|
|
const url = await server(t, (_request, response) => { publicReads++; response.end(CONTENT); });
|
|
const entry = item(url, [{ id: '9007199254740993' }, { id: '9007199254740997' }]);
|
|
const db = { execute: mock.fn(async () => [[stored(entry)]]) };
|
|
const [result] = await verifySwitchedBatch(db, CONFIG, [entry]);
|
|
assert.ok(result.error);
|
|
assert.equal(publicReads, 0);
|
|
});
|
|
|
|
test('switched public corruption excludes only its own mapping and keeps input order', async t => {
|
|
const url = await server(t, (request, response) => response.end(request.url === '/bad' ? CORRUPT : CONTENT));
|
|
const bad = item(url + '/bad', []);
|
|
const good = item(url + '/good', []);
|
|
const results = await verifySwitchedBatch({}, CONFIG, [bad, good]);
|
|
assert.equal(results[0].item, bad);
|
|
assert.match(results[0].error.message, /SHA-256/);
|
|
assert.deepEqual(results[1], { item: good });
|
|
assert.deepEqual(results.filter(result => !result.error).map(result => result.item.url), [good.url]);
|
|
});
|
|
|
|
test('metadata query errors reject before network validation starts', async t => {
|
|
let publicReads = 0;
|
|
const url = await server(t, (_request, response) => { publicReads++; response.end(CONTENT); });
|
|
const failure = new Error('metadata unavailable');
|
|
const db = { execute: mock.fn(async () => { throw failure; }) };
|
|
await assert.rejects(verifySwitchedBatch(db, CONFIG, [item(url)]), error => error === failure);
|
|
assert.equal(publicReads, 0);
|
|
});
|
|
|
|
test('final validation is capped at four active requests and settles only after every started request', { timeout: 4000 }, async t => {
|
|
const responses = new Map();
|
|
let active = 0, maximum = 0, settled = false;
|
|
const url = await server(t, (request, response) => {
|
|
active++;
|
|
maximum = Math.max(maximum, active);
|
|
response.once('finish', () => { active--; });
|
|
responses.set(Number(request.url.slice(1)), response);
|
|
});
|
|
const batch = Array.from({ length: 6 }, (_, index) => item(url + '/' + index, []));
|
|
const pending = verifySwitchedBatch({}, CONFIG, batch).then(results => { settled = true; return results; });
|
|
const until = async predicate => {
|
|
const deadline = Date.now() + 1500;
|
|
while (!predicate()) {
|
|
if (Date.now() > deadline) assert.fail('validation did not reach the expected request state');
|
|
await setImmediate();
|
|
}
|
|
};
|
|
await until(() => responses.size === 4);
|
|
assert.equal(active, 4);
|
|
assert.equal(settled, false);
|
|
// Per-item failures are collected, so remaining items must still be checked.
|
|
responses.get(0).end(CORRUPT);
|
|
responses.get(1).end(CONTENT);
|
|
await until(() => responses.size === 6);
|
|
assert.equal(active, 4);
|
|
assert.equal(settled, false);
|
|
responses.get(2).end(CONTENT);
|
|
responses.get(3).end(CONTENT);
|
|
responses.get(4).end(CONTENT);
|
|
await setImmediate();
|
|
assert.equal(settled, false);
|
|
responses.get(5).end(CONTENT);
|
|
const results = await pending;
|
|
assert.equal(maximum, 4);
|
|
assert.equal(active, 0);
|
|
assert.equal(settled, true);
|
|
assert.equal(results.length, 6);
|
|
assert.match(results[0].error.message, /SHA-256/);
|
|
assert.deepEqual(results.slice(1), batch.slice(1).map(entry => ({ item: entry })));
|
|
});
|
|
|
|
test('empty switched batches perform no database or network work', async () => {
|
|
const db = { execute: mock.fn(async () => { throw new Error('must not query'); }) };
|
|
assert.deepEqual(await verifySwitchedBatch(db, CONFIG, []), []);
|
|
assert.equal(db.execute.mock.callCount(), 0);
|
|
});
|
|
|
|
test('more than 1000 file records are loaded in bounded sequential queries before public validation', async t => {
|
|
let completedQueries = 0, activeQueries = 0, maximumQueries = 0, publicReads = 0;
|
|
const url = await server(t, (_request, response) => {
|
|
publicReads++;
|
|
assert.equal(completedQueries, 2);
|
|
assert.equal(activeQueries, 0);
|
|
response.end(CONTENT);
|
|
});
|
|
const files = Array.from({ length: 1001 }, (_, index) => ({ id: String(9007199254740993n + BigInt(index)) }));
|
|
const entry = item(url, files);
|
|
const seenIds = [];
|
|
const querySizes = [];
|
|
const db = { execute: mock.fn(async (_sql, params) => {
|
|
maximumQueries = Math.max(maximumQueries, ++activeQueries);
|
|
assert.ok(params.length <= 1000);
|
|
assert.equal(publicReads, 0);
|
|
querySizes.push(params.length);
|
|
seenIds.push(...params);
|
|
await setImmediate();
|
|
completedQueries++;
|
|
activeQueries--;
|
|
return [params.map(id => ({ ...stored(entry), id })).reverse()];
|
|
}) };
|
|
assert.deepEqual(await verifySwitchedBatch(db, CONFIG, [entry]), [{ item: entry }]);
|
|
assert.deepEqual(querySizes, [1000, 1]);
|
|
assert.deepEqual(seenIds, files.map(file => file.id));
|
|
assert.equal(maximumQueries, 1);
|
|
assert.equal(db.execute.mock.callCount(), 2);
|
|
assert.equal(publicReads, 1);
|
|
});
|