270 lines
8.0 KiB
TypeScript
270 lines
8.0 KiB
TypeScript
import { describe, expect, it, vi } from 'vitest';
|
|
import {
|
|
MediaJobQueue,
|
|
MemoryJobMetadataStore,
|
|
ObjectUrlRegistry,
|
|
parseFFmpegProgress,
|
|
progressUpdateFromFFmpeg,
|
|
weightedJobProgress,
|
|
type JobExecutionContext,
|
|
type JobExecutionResult,
|
|
type MediaJob,
|
|
type MediaJobDefinition,
|
|
type MediaJobExecutor,
|
|
} from '../../src/jobs';
|
|
import { freezeCommandPlan } from '../../src/commands';
|
|
|
|
function definition(id: string, stepIds = ['encode']): MediaJobDefinition {
|
|
return {
|
|
id,
|
|
operation: 'test',
|
|
steps: stepIds.map((stepId) => ({
|
|
id: stepId,
|
|
name: stepId,
|
|
weight: stepId === 'read' ? 1 : 9,
|
|
plan: freezeCommandPlan({
|
|
id: `${id}:${stepId}`,
|
|
operation: stepId,
|
|
inputs: [],
|
|
temporaryFiles: [],
|
|
args: ['-version'],
|
|
outputs: [],
|
|
requiredCapabilities: {},
|
|
diagnostics: [],
|
|
}),
|
|
})),
|
|
};
|
|
}
|
|
|
|
class ControlledExecutor implements MediaJobExecutor {
|
|
readonly calls: string[] = [];
|
|
readonly cancelCurrentJob = vi.fn(async () => undefined);
|
|
readonly recoverAfterCancellation = vi.fn(async () => undefined);
|
|
readonly cleanup = vi.fn(async () => undefined);
|
|
#gates: (() => void)[] = [];
|
|
failNext?: Error;
|
|
|
|
execute(
|
|
plan: { readonly id: string },
|
|
context: JobExecutionContext
|
|
): Promise<JobExecutionResult> {
|
|
this.calls.push(plan.id);
|
|
context.onProgress({ progress: 0.5 });
|
|
if (this.failNext) {
|
|
const error = this.failNext;
|
|
this.failNext = undefined;
|
|
return Promise.reject(error);
|
|
}
|
|
return new Promise((resolve, reject) => {
|
|
const release = () => {
|
|
context.signal.removeEventListener('abort', abort);
|
|
resolve({ outputs: new Map() });
|
|
};
|
|
const abort = () => {
|
|
const index = this.#gates.indexOf(release);
|
|
if (index >= 0) {
|
|
this.#gates.splice(index, 1);
|
|
}
|
|
reject(new DOMException('cancelled', 'AbortError'));
|
|
};
|
|
context.signal.addEventListener('abort', abort, { once: true });
|
|
this.#gates.push(release);
|
|
});
|
|
}
|
|
|
|
release(): void {
|
|
this.#gates.shift()?.();
|
|
}
|
|
}
|
|
|
|
describe('progress parsing', () => {
|
|
it('parses FFmpeg machine progress and caps completion until output read', () => {
|
|
const parsed = parseFFmpegProgress(
|
|
'frame=25\nfps=12.5\nout_time_us=5000000\nspeed=1.5x\nprogress=continue\n'
|
|
);
|
|
expect(parsed).toEqual({
|
|
frame: 25,
|
|
fps: 12.5,
|
|
outTimeSeconds: 5,
|
|
speed: 1.5,
|
|
terminal: false,
|
|
});
|
|
expect(progressUpdateFromFFmpeg(parsed, 10)).toMatchObject({
|
|
progress: 0.5,
|
|
});
|
|
expect(
|
|
progressUpdateFromFFmpeg(
|
|
parseFFmpegProgress('progress=end\nout_time_us=10000000'),
|
|
10
|
|
).progress
|
|
).toBe(0.99);
|
|
});
|
|
|
|
it('combines weighted multi-step progress below 100 percent', () => {
|
|
expect(
|
|
weightedJobProgress([
|
|
{
|
|
id: 'analysis',
|
|
name: 'Analysis',
|
|
weight: 1,
|
|
status: 'completed',
|
|
progress: 1,
|
|
planId: 'one',
|
|
},
|
|
{
|
|
id: 'encode',
|
|
name: 'Encoding',
|
|
weight: 9,
|
|
status: 'running',
|
|
progress: 0.5,
|
|
planId: 'two',
|
|
},
|
|
])
|
|
).toBe(0.55);
|
|
});
|
|
});
|
|
|
|
describe('sequential durable job queue', () => {
|
|
it('runs only one command at a time and reaches 100% after cleanup', async () => {
|
|
const executor = new ControlledExecutor();
|
|
const store = new MemoryJobMetadataStore();
|
|
const queue = new MediaJobQueue({ executor, store });
|
|
await queue.initialize();
|
|
await queue.enqueue(definition('one'));
|
|
await queue.enqueue(definition('two'));
|
|
await waitUntil(() => executor.calls.length === 1);
|
|
expect(executor.calls).toEqual(['one:encode']);
|
|
expect(queue.get('one')?.progress).toBe(0.5);
|
|
executor.release();
|
|
await waitUntil(() => executor.calls.length === 2);
|
|
expect(queue.get('one')).toMatchObject({
|
|
status: 'completed',
|
|
progress: 1,
|
|
});
|
|
executor.release();
|
|
await queue.waitForIdle();
|
|
expect(queue.get('two')).toMatchObject({
|
|
status: 'completed',
|
|
progress: 1,
|
|
});
|
|
expect(queue.getResults('two')).toHaveLength(1);
|
|
expect(queue.releaseResults('two')).toBe(true);
|
|
expect(queue.getResults('two')).toBeUndefined();
|
|
expect((await store.list()).map((job) => job.status)).toEqual([
|
|
'completed',
|
|
'completed',
|
|
]);
|
|
});
|
|
|
|
it('terminates processing, recovers the engine, and runs the next job', async () => {
|
|
const executor = new ControlledExecutor();
|
|
const queue = new MediaJobQueue({
|
|
executor,
|
|
store: new MemoryJobMetadataStore(),
|
|
});
|
|
await queue.initialize();
|
|
await queue.enqueue(definition('cancel-me'));
|
|
await queue.enqueue(definition('after'));
|
|
await waitUntil(() => queue.get('cancel-me')?.status === 'running');
|
|
await queue.cancel('cancel-me');
|
|
expect(executor.cancelCurrentJob).toHaveBeenCalledOnce();
|
|
expect(executor.recoverAfterCancellation).toHaveBeenCalledOnce();
|
|
expect(queue.get('cancel-me')?.status).toBe('cancelled');
|
|
await waitUntil(() => executor.calls.includes('after:encode'));
|
|
executor.release();
|
|
await queue.waitForIdle();
|
|
expect(queue.get('after')?.status).toBe('completed');
|
|
});
|
|
|
|
it('marks failures, cleans up, and retries with the retained definition', async () => {
|
|
const executor = new ControlledExecutor();
|
|
executor.failNext = new Error('encoder failed');
|
|
const queue = new MediaJobQueue({
|
|
executor,
|
|
store: new MemoryJobMetadataStore(),
|
|
});
|
|
await queue.initialize();
|
|
await queue.enqueue(definition('retry'));
|
|
await queue.waitForIdle();
|
|
expect(queue.get('retry')).toMatchObject({
|
|
status: 'failed',
|
|
error: { code: 'execution-failed', recoverable: true },
|
|
});
|
|
await queue.retry('retry');
|
|
await waitUntil(() => executor.calls.length === 2);
|
|
executor.release();
|
|
await queue.waitForIdle();
|
|
expect(queue.get('retry')).toMatchObject({
|
|
status: 'completed',
|
|
retryCount: 1,
|
|
});
|
|
});
|
|
|
|
it('recovers interrupted durable metadata as an explicit failed job', async () => {
|
|
const store = new MemoryJobMetadataStore();
|
|
const interrupted: MediaJob = {
|
|
id: 'old',
|
|
operation: 'convert',
|
|
status: 'running',
|
|
steps: [
|
|
{
|
|
id: 'encode',
|
|
name: 'Encode',
|
|
weight: 1,
|
|
status: 'running',
|
|
progress: 0.4,
|
|
planId: 'old:encode',
|
|
},
|
|
],
|
|
progress: 0.4,
|
|
createdAt: '2026-01-01T00:00:00.000Z',
|
|
updatedAt: '2026-01-01T00:01:00.000Z',
|
|
startedAt: '2026-01-01T00:00:01.000Z',
|
|
retryCount: 0,
|
|
};
|
|
await store.put(interrupted);
|
|
const queue = new MediaJobQueue({
|
|
executor: new ControlledExecutor(),
|
|
store,
|
|
});
|
|
await queue.initialize();
|
|
expect(queue.get('old')).toMatchObject({
|
|
status: 'failed',
|
|
recoveryNote: expect.stringContaining('previous browser session'),
|
|
error: { code: 'session-interrupted', recoverable: true },
|
|
});
|
|
});
|
|
});
|
|
|
|
describe('object URL cleanup', () => {
|
|
it('revokes replaced and cleared result URLs', () => {
|
|
let index = 0;
|
|
const revoke = vi.fn();
|
|
const registry = new ObjectUrlRegistry({
|
|
createObjectURL: () => `blob:${index++}`,
|
|
revokeObjectURL: revoke,
|
|
});
|
|
expect(registry.replace('output', new Blob(['one']))).toBe('blob:0');
|
|
expect(registry.replace('output', new Blob(['two']))).toBe('blob:1');
|
|
registry.replace('preview', new Blob(['three']));
|
|
expect(revoke).toHaveBeenCalledWith('blob:0');
|
|
registry.clear();
|
|
expect(revoke).toHaveBeenCalledWith('blob:1');
|
|
expect(revoke).toHaveBeenCalledWith('blob:2');
|
|
expect(registry.size).toBe(0);
|
|
});
|
|
});
|
|
|
|
async function waitUntil(
|
|
predicate: () => boolean,
|
|
timeoutMs = 2_000
|
|
): Promise<void> {
|
|
const started = Date.now();
|
|
while (!predicate()) {
|
|
if (Date.now() - started > timeoutMs) {
|
|
throw new Error('Timed out waiting for queue state');
|
|
}
|
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
|
}
|
|
}
|