huangzd1997
93943bb51a
in-flight 期间不叠加 timer、不并发请求；完成按退避续轮； 重复 ensure(true) 不叠加；add 期间请求含新任务；失败释放锁。 共 8 个测试。
246 lines
7.4 KiB
TypeScript
246 lines
7.4 KiB
TypeScript
import { test } from 'node:test'
|
||
import assert from 'node:assert/strict'
|
||
import { useTaskProgressLoop } from '../src/shared/composables/useTaskProgressLoop.ts'
|
||
import { getTaskPollBackoffMs } from '../src/shared/task-progress-config.ts'
|
||
|
||
function setupWindow() {
|
||
const storage = new Map<string, string>()
|
||
;(globalThis as Record<string, unknown>).window = {
|
||
localStorage: {
|
||
getItem: (k: string) => storage.get(k) ?? null,
|
||
setItem: (k: string, v: string) => { storage.set(k, v) },
|
||
removeItem: (k: string) => { storage.delete(k) },
|
||
},
|
||
setTimeout: (fn: () => void, ms: number) => globalThis.setTimeout(fn, ms),
|
||
clearTimeout: (id: unknown) => globalThis.clearTimeout(id as number),
|
||
setInterval: (fn: () => void, ms: number) => globalThis.setInterval(fn, ms),
|
||
clearInterval: (id: unknown) => globalThis.clearInterval(id as number),
|
||
}
|
||
}
|
||
|
||
const tick = (ms = 0) => new Promise<void>((r) => globalThis.setTimeout(r, ms))
|
||
|
||
async function waitFor(cond: () => boolean, timeoutMs: number): Promise<boolean> {
|
||
const deadline = Date.now() + timeoutMs
|
||
while (!cond() && Date.now() < deadline) {
|
||
await tick(10)
|
||
}
|
||
return cond()
|
||
}
|
||
|
||
let scopeCounter = 0
|
||
|
||
interface GatedHarness {
|
||
loop: ReturnType<typeof useTaskProgressLoop<{ taskId?: number; status?: string }>>
|
||
fetchCount: () => number
|
||
maxConcurrent: () => number
|
||
fetchIdsCalls: () => number[][]
|
||
release: (index?: number) => void
|
||
releaseAll: () => void
|
||
}
|
||
|
||
/** 每个请求都阻塞在 gate 上，直到测试手动释放；统计并发峰值 */
|
||
async function makeGatedLoop(): Promise<GatedHarness> {
|
||
const gates: Array<() => void> = []
|
||
const fetchCalls: number[][] = []
|
||
let count = 0
|
||
let active = 0
|
||
let peak = 0
|
||
const loop = useTaskProgressLoop<{ taskId?: number; status?: string }>({
|
||
scope: `inflight-${scopeCounter++}`,
|
||
fetchProgress: async (ids) => {
|
||
count += 1
|
||
active += 1
|
||
peak = Math.max(peak, active)
|
||
fetchCalls.push(ids.slice())
|
||
await new Promise<void>((r) => { gates.push(r) })
|
||
active -= 1
|
||
return { items: [] }
|
||
},
|
||
extractTaskId: (d) => d?.taskId,
|
||
extractStatus: (d) => d?.status,
|
||
getIntervalMs: () => 60000,
|
||
})
|
||
return {
|
||
loop,
|
||
fetchCount: () => count,
|
||
maxConcurrent: () => peak,
|
||
fetchIdsCalls: () => fetchCalls,
|
||
release: (i = 0) => { gates[i]?.() },
|
||
releaseAll: () => { for (const g of gates) g() },
|
||
}
|
||
}
|
||
|
||
test('test_in_flight_no_second_request', async () => {
|
||
setupWindow()
|
||
const h = await makeGatedLoop()
|
||
h.loop.reset([1])
|
||
await tick(0)
|
||
assert.equal(h.fetchCount(), 1, '首轮请求已进入 in-flight')
|
||
h.loop.ensure(true)
|
||
await tick(60)
|
||
assert.equal(h.fetchCount(), 1, 'in-flight 期间不发起第二个请求')
|
||
h.release(0)
|
||
assert.ok(await waitFor(() => h.fetchCount() >= 2, 2000), '上一次完成后按退避继续')
|
||
h.releaseAll()
|
||
h.loop.dispose()
|
||
})
|
||
|
||
test('test_in_flight_queues_after', async () => {
|
||
setupWindow()
|
||
let count = 0
|
||
let active = 0
|
||
let peak = 0
|
||
const loop = useTaskProgressLoop<{ taskId?: number; status?: string }>({
|
||
scope: `inflight-seq-${scopeCounter++}`,
|
||
fetchProgress: async () => {
|
||
count += 1
|
||
active += 1
|
||
peak = Math.max(peak, active)
|
||
await tick(30)
|
||
active -= 1
|
||
return { items: [] }
|
||
},
|
||
extractTaskId: (d) => d?.taskId,
|
||
extractStatus: (d) => d?.status,
|
||
getIntervalMs: () => 40,
|
||
})
|
||
loop.reset([1])
|
||
assert.ok(await waitFor(() => count >= 2, 2000), '请求完成后自动进入下一轮')
|
||
assert.equal(peak, 1, '全程无并发')
|
||
loop.dispose()
|
||
})
|
||
|
||
test('test_in_flight_backoff_wait', async () => {
|
||
setupWindow()
|
||
const h = await makeGatedLoop()
|
||
h.loop.reset([1])
|
||
await tick(0)
|
||
assert.equal(h.fetchCount(), 1)
|
||
const t0 = Date.now()
|
||
h.loop.ensure(true)
|
||
h.release(0)
|
||
assert.ok(await waitFor(() => h.fetchCount() >= 2, 3000), 'in-flight 结束后按退避重试')
|
||
const elapsed = Date.now() - t0
|
||
assert.ok(
|
||
elapsed >= getTaskPollBackoffMs(0) - 100,
|
||
`重试等待 >= 退避时长 ${getTaskPollBackoffMs(0)}ms（实际 ${elapsed}ms）`,
|
||
)
|
||
h.releaseAll()
|
||
h.loop.dispose()
|
||
})
|
||
|
||
test('test_in_flight_timer_single', async () => {
|
||
setupWindow()
|
||
const h = await makeGatedLoop()
|
||
h.loop.reset([1])
|
||
await tick(0)
|
||
h.loop.ensure(true)
|
||
h.loop.ensure(true)
|
||
h.release(0)
|
||
assert.ok(await waitFor(() => h.fetchCount() >= 2, 2000), '重复 ensure 只保留一个退避 timer')
|
||
await tick(800)
|
||
assert.equal(h.fetchCount(), 2, '无叠加 timer 引发的第三次请求')
|
||
h.releaseAll()
|
||
h.loop.dispose()
|
||
})
|
||
|
||
test('test_in_flight_add_during', async () => {
|
||
setupWindow()
|
||
const h = await makeGatedLoop()
|
||
h.loop.reset([1])
|
||
await tick(0)
|
||
assert.equal(h.fetchCount(), 1)
|
||
h.loop.add(2)
|
||
assert.deepEqual(h.loop.taskIds.value, [1, 2], 'in-flight 期间 add 任务入集合')
|
||
await tick(60)
|
||
assert.equal(h.fetchCount(), 1, 'add 不触发并发请求')
|
||
h.release(0)
|
||
assert.ok(await waitFor(() => h.fetchCount() >= 2, 2000), '退避后轮询新任务集合')
|
||
assert.deepEqual(h.fetchIdsCalls()[1], [1, 2], '下一轮请求包含新任务')
|
||
assert.equal(h.maxConcurrent(), 1)
|
||
h.releaseAll()
|
||
h.loop.dispose()
|
||
})
|
||
|
||
test('test_in_flight_error_releases', async () => {
|
||
setupWindow()
|
||
let fail = true
|
||
const errors: unknown[] = []
|
||
const loop = useTaskProgressLoop<{ taskId?: number; status?: string }>({
|
||
scope: `inflight-err-${scopeCounter++}`,
|
||
fetchProgress: async () => {
|
||
if (fail) throw new Error('boom')
|
||
return { items: [{ taskId: 1, status: 'RUNNING' }] }
|
||
},
|
||
extractTaskId: (d) => d?.taskId,
|
||
extractStatus: (d) => d?.status,
|
||
getIntervalMs: () => 60000,
|
||
onError: (e) => { errors.push(e) },
|
||
})
|
||
loop.reset([1])
|
||
await tick(0)
|
||
assert.equal(errors.length, 1, '首轮失败上报 onError')
|
||
assert.equal(loop.inFlight.value, false, '失败后 in-flight 锁释放')
|
||
fail = false
|
||
await loop.refreshOnce()
|
||
assert.equal(loop.taskStatuses.value[1], 'RUNNING', '锁释放后可正常轮询')
|
||
loop.dispose()
|
||
})
|
||
|
||
test('test_in_flight_immediate_repeat', async () => {
|
||
setupWindow()
|
||
let count = 0
|
||
let active = 0
|
||
let peak = 0
|
||
const loop = useTaskProgressLoop<{ taskId?: number; status?: string }>({
|
||
scope: `inflight-rep-${scopeCounter++}`,
|
||
fetchProgress: async () => {
|
||
count += 1
|
||
active += 1
|
||
peak = Math.max(peak, active)
|
||
await tick(30)
|
||
active -= 1
|
||
return { items: [] }
|
||
},
|
||
extractTaskId: (d) => d?.taskId,
|
||
extractStatus: (d) => d?.status,
|
||
getIntervalMs: () => 60000,
|
||
})
|
||
loop.reset([1])
|
||
await tick(0)
|
||
assert.equal(count, 1)
|
||
loop.ensure(true)
|
||
loop.ensure(true)
|
||
await tick(40)
|
||
assert.equal(count, 1, '请求未完成时重复 ensure(true) 不并发')
|
||
assert.equal(peak, 1)
|
||
assert.ok(await waitFor(() => count >= 2, 2000), '完成后退避续轮')
|
||
loop.dispose()
|
||
})
|
||
|
||
test('test_in_flight_long_request', async () => {
|
||
setupWindow()
|
||
let count = 0
|
||
let active = 0
|
||
let peak = 0
|
||
const loop = useTaskProgressLoop<{ taskId?: number; status?: string }>({
|
||
scope: `inflight-long-${scopeCounter++}`,
|
||
fetchProgress: async () => {
|
||
count += 1
|
||
active += 1
|
||
peak = Math.max(peak, active)
|
||
await tick(50)
|
||
active -= 1
|
||
return { items: [] }
|
||
},
|
||
extractTaskId: (d) => d?.taskId,
|
||
extractStatus: (d) => d?.status,
|
||
getIntervalMs: () => 25,
|
||
})
|
||
loop.reset([1])
|
||
assert.ok(await waitFor(() => count >= 3, 4000), '长请求期间多次续轮')
|
||
assert.equal(peak, 1, '长请求期间始终单请求')
|
||
loop.dispose()
|
||
})
|