125 lines
4.5 KiB
JavaScript
125 lines
4.5 KiB
JavaScript
'use strict';
|
|
|
|
const assert = require('assert');
|
|
const { runConcurrentItems } = require('../src/utils/concurrent_runner');
|
|
|
|
/**
|
|
* 让并发任务主动让出事件循环，便于测试真实的滑动窗口并发度。
|
|
*/
|
|
function yieldEventLoop() {
|
|
return new Promise(resolve => setImmediate(resolve));
|
|
}
|
|
|
|
(async () => {
|
|
const accounts = Array.from({ length: 545 }, (_, index) => ({
|
|
email: `user${index}@example.com`,
|
|
}));
|
|
const processed = new Set();
|
|
const progress = [];
|
|
let active = 0;
|
|
let maxActive = 0;
|
|
|
|
const batch = await runConcurrentItems(accounts, async (account, index) => {
|
|
processed.add(account.email);
|
|
active += 1;
|
|
maxActive = Math.max(maxActive, active);
|
|
try {
|
|
await yieldEventLoop();
|
|
if (index === 109) throw new Error('DB write timeout: updateAccountStatusReason');
|
|
return { email: account.email, ok: true };
|
|
} finally {
|
|
active -= 1;
|
|
}
|
|
}, {
|
|
concurrency: 50,
|
|
onError: (error, account) => ({
|
|
email: account.email,
|
|
ok: false,
|
|
code: 'ACCOUNT_TASK_ERROR',
|
|
msg: error.message,
|
|
}),
|
|
onProgress: ({ completed }) => {
|
|
progress.push(completed);
|
|
if (completed === 110) throw new Error('模拟前端进度通道短暂异常');
|
|
},
|
|
});
|
|
|
|
assert.strictEqual(processed.size, 545, '第 110 个账号失败后仍应调度完全部 545 个账号');
|
|
assert.strictEqual(batch.completed, 545, '完成计数必须包含成功和失败账号');
|
|
assert.strictEqual(batch.results.filter(Boolean).length, 545, '每个账号都必须留下明确结果');
|
|
assert.strictEqual(batch.results[109].ok, false, '单账号写库超时应保留为该账号失败');
|
|
assert.strictEqual(progress.at(-1), 545, '最终进度必须到达批次总数');
|
|
assert.strictEqual(maxActive, 50, '业务并发仍应按用户设置跑满 50');
|
|
|
|
let paused = true;
|
|
let pauseWaiters = [];
|
|
const pauseStarted = [];
|
|
const pausedBatchPromise = runConcurrentItems([1, 2, 3, 4], async item => {
|
|
pauseStarted.push(item);
|
|
await yieldEventLoop();
|
|
return item;
|
|
}, {
|
|
concurrency: 2,
|
|
waitUntilReady: async () => {
|
|
while (paused) await new Promise(resolve => pauseWaiters.push(resolve));
|
|
},
|
|
});
|
|
await new Promise(resolve => setTimeout(resolve, 20));
|
|
assert.deepStrictEqual(pauseStarted, [], '暂停期间不得领取任何新项目');
|
|
paused = false;
|
|
const releasing = pauseWaiters;
|
|
pauseWaiters = [];
|
|
releasing.forEach(resolve => resolve());
|
|
const resumedBatch = await pausedBatchPromise;
|
|
assert.strictEqual(resumedBatch.completed, 4, '继续后必须处理完全部剩余项目');
|
|
assert.deepStrictEqual(pauseStarted.sort(), [1, 2, 3, 4], '继续后不得遗漏项目');
|
|
|
|
const deferredRuns = [];
|
|
const attempts = new Map();
|
|
const deferredBatch = await runConcurrentItems(['sleeping', 'next-a', 'next-b'], async item => {
|
|
deferredRuns.push(item);
|
|
const attempt = (attempts.get(item) || 0) + 1;
|
|
attempts.set(item, attempt);
|
|
if (item === 'sleeping' && attempt === 1) {
|
|
return { defer: true, delayMs: 40 };
|
|
}
|
|
await yieldEventLoop();
|
|
return { item, attempt };
|
|
}, {
|
|
concurrency: 1,
|
|
getDeferDelay: result => (result && result.defer ? result.delayMs : 0),
|
|
});
|
|
assert.deepStrictEqual(
|
|
deferredRuns,
|
|
['sleeping', 'next-a', 'next-b', 'sleeping'],
|
|
'延迟项目等待期间必须让出唯一并发槽，让后续项目先完成',
|
|
);
|
|
assert.strictEqual(deferredBatch.completed, 3, '延迟重试不应重复累计完成进度');
|
|
assert.strictEqual(deferredBatch.results[0].attempt, 2, '延迟项目到期后应在原索引写入最终结果');
|
|
|
|
const heldSlotRuns = [];
|
|
const heldSlotAttempts = new Map();
|
|
const heldSlotBatch = await runConcurrentItems(['sleeping', 'next'], async item => {
|
|
heldSlotRuns.push(item);
|
|
const attempt = (heldSlotAttempts.get(item) || 0) + 1;
|
|
heldSlotAttempts.set(item, attempt);
|
|
if (item === 'sleeping' && attempt === 1) return { defer: true, delayMs: 20 };
|
|
return { item, attempt };
|
|
}, {
|
|
concurrency: 1,
|
|
deferReleasesSlot: false,
|
|
getDeferDelay: result => (result && result.defer ? result.delayMs : 0),
|
|
});
|
|
assert.deepStrictEqual(
|
|
heldSlotRuns,
|
|
['sleeping', 'sleeping', 'next'],
|
|
'临时比特浏览器任务延迟等待时必须占住串行槽，删除后才能轮到下一账号',
|
|
);
|
|
assert.strictEqual(heldSlotBatch.completed, 2, '占住串行槽时仍应完成全部项目');
|
|
|
|
console.log('concurrent_runner_resilience.test.js: PASS');
|
|
})().catch(error => {
|
|
console.error(error);
|
|
process.exitCode = 1;
|
|
});
|