compute: reuse shared timeout helpers across worker and local
This commit is contained in:
parent
d70c6aa5c6
commit
c9dad9c23c
3 changed files with 78 additions and 136 deletions
|
|
@ -11,6 +11,12 @@ export type ComputeTimeoutConfig = {
|
|||
};
|
||||
|
||||
export type ComputeOperationKind = 'whisper_align' | 'pdf_layout';
|
||||
export type IdleTimeoutAndHardCapInput<T> = {
|
||||
run: (touchProgress: () => void) => Promise<T>;
|
||||
idleTimeoutMs: number;
|
||||
hardCapMs: number;
|
||||
label: string;
|
||||
};
|
||||
|
||||
function readPositiveIntEnv(name: string, fallback: number): number {
|
||||
const raw = process.env[name]?.trim();
|
||||
|
|
@ -37,3 +43,65 @@ export function getWorkerClientWaitTimeoutMs(kind: ComputeOperationKind): number
|
|||
const timeoutMs = kind === 'pdf_layout' ? config.pdfTimeoutMs : config.whisperTimeoutMs;
|
||||
return Math.max(DEFAULT_WORKER_WAIT_MIN_MS, timeoutMs + DEFAULT_WORKER_WAIT_BUFFER_MS);
|
||||
}
|
||||
|
||||
export async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
|
||||
let timer: NodeJS.Timeout | null = null;
|
||||
try {
|
||||
return await Promise.race([
|
||||
promise,
|
||||
new Promise<T>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error(`${label} timed out after ${timeoutMs}ms`)), timeoutMs);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
export async function withIdleTimeoutAndHardCap<T>(input: IdleTimeoutAndHardCapInput<T>): Promise<T> {
|
||||
let idleTimer: NodeJS.Timeout | null = null;
|
||||
let hardCapTimer: NodeJS.Timeout | null = null;
|
||||
let settled = false;
|
||||
let rejectTimeout!: (reason: unknown) => void;
|
||||
|
||||
const clearTimers = () => {
|
||||
if (idleTimer) {
|
||||
clearTimeout(idleTimer);
|
||||
idleTimer = null;
|
||||
}
|
||||
if (hardCapTimer) {
|
||||
clearTimeout(hardCapTimer);
|
||||
hardCapTimer = null;
|
||||
}
|
||||
};
|
||||
|
||||
const failTimeout = (kind: 'idle' | 'hard cap', timeoutMs: number) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimers();
|
||||
rejectTimeout(new Error(`${input.label} ${kind} timed out after ${timeoutMs}ms`));
|
||||
};
|
||||
|
||||
const touchProgress = () => {
|
||||
if (settled) return;
|
||||
if (idleTimer) clearTimeout(idleTimer);
|
||||
idleTimer = setTimeout(() => failTimeout('idle', input.idleTimeoutMs), input.idleTimeoutMs);
|
||||
};
|
||||
|
||||
const timeoutPromise = new Promise<never>((_, reject) => {
|
||||
rejectTimeout = reject;
|
||||
hardCapTimer = setTimeout(() => failTimeout('hard cap', input.hardCapMs), input.hardCapMs);
|
||||
touchProgress();
|
||||
});
|
||||
|
||||
try {
|
||||
const result = await Promise.race([input.run(touchProgress), timeoutPromise]);
|
||||
settled = true;
|
||||
clearTimers();
|
||||
return result as T;
|
||||
} catch (error) {
|
||||
settled = true;
|
||||
clearTimers();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -27,7 +27,11 @@ import {
|
|||
runPdfLayoutFromPdfBuffer,
|
||||
runWhisperAlignmentFromAudioBuffer,
|
||||
} from '@openreader/compute-core/local-runtime';
|
||||
import { getComputeTimeoutConfig } from '@openreader/compute-core/runtime/timeout-config';
|
||||
import {
|
||||
getComputeTimeoutConfig,
|
||||
withIdleTimeoutAndHardCap,
|
||||
withTimeout,
|
||||
} from '@openreader/compute-core/runtime/timeout-config';
|
||||
import {
|
||||
type PdfLayoutJobRequest,
|
||||
type PdfLayoutJobResult,
|
||||
|
|
@ -215,73 +219,6 @@ function toArrayBuffer(bytes: Uint8Array): ArrayBuffer {
|
|||
return copy.buffer;
|
||||
}
|
||||
|
||||
async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
|
||||
let timer: NodeJS.Timeout | null = null;
|
||||
try {
|
||||
return await Promise.race([
|
||||
promise,
|
||||
new Promise<T>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error(`${label} timed out after ${timeoutMs}ms`)), timeoutMs);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
async function withIdleTimeoutAndHardCap<T>(input: {
|
||||
run: (touchProgress: () => void) => Promise<T>;
|
||||
idleTimeoutMs: number;
|
||||
hardCapMs: number;
|
||||
label: string;
|
||||
}): Promise<T> {
|
||||
let idleTimer: NodeJS.Timeout | null = null;
|
||||
let hardCapTimer: NodeJS.Timeout | null = null;
|
||||
let settled = false;
|
||||
let rejectTimeout!: (reason: unknown) => void;
|
||||
|
||||
const clearTimers = () => {
|
||||
if (idleTimer) {
|
||||
clearTimeout(idleTimer);
|
||||
idleTimer = null;
|
||||
}
|
||||
if (hardCapTimer) {
|
||||
clearTimeout(hardCapTimer);
|
||||
hardCapTimer = null;
|
||||
}
|
||||
};
|
||||
|
||||
const failTimeout = (kind: 'idle' | 'hard cap', timeoutMs: number) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimers();
|
||||
rejectTimeout(new Error(`${input.label} ${kind} timed out after ${timeoutMs}ms`));
|
||||
};
|
||||
|
||||
const touchProgress = () => {
|
||||
if (settled) return;
|
||||
if (idleTimer) clearTimeout(idleTimer);
|
||||
idleTimer = setTimeout(() => failTimeout('idle', input.idleTimeoutMs), input.idleTimeoutMs);
|
||||
};
|
||||
|
||||
const timeoutPromise = new Promise<never>((_, reject) => {
|
||||
rejectTimeout = reject;
|
||||
hardCapTimer = setTimeout(() => failTimeout('hard cap', input.hardCapMs), input.hardCapMs);
|
||||
touchProgress();
|
||||
});
|
||||
|
||||
try {
|
||||
const result = await Promise.race([input.run(touchProgress), timeoutPromise]);
|
||||
settled = true;
|
||||
clearTimers();
|
||||
return result as T;
|
||||
} catch (error) {
|
||||
settled = true;
|
||||
clearTimers();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
function isAuthed(request: FastifyRequest, expectedToken: string): boolean {
|
||||
const auth = request.headers.authorization;
|
||||
if (!auth?.startsWith('Bearer ')) return false;
|
||||
|
|
|
|||
|
|
@ -2,80 +2,17 @@ import type { ComputeBackend, PdfLayoutInput, WhisperAlignInput, WhisperAlignRes
|
|||
import { LOCAL_COMPUTE_LIMITER } from '@/lib/server/compute/concurrency-limiter';
|
||||
import { getDocumentBlob } from '@/lib/server/documents/blobstore';
|
||||
import { getTtsSegmentAudioObject } from '@/lib/server/tts/segments-blobstore';
|
||||
import { getComputeTimeoutConfig } from '@openreader/compute-core/runtime/timeout-config';
|
||||
import {
|
||||
getComputeTimeoutConfig,
|
||||
withIdleTimeoutAndHardCap,
|
||||
withTimeout,
|
||||
} from '@openreader/compute-core/runtime/timeout-config';
|
||||
import {
|
||||
runPdfLayoutFromPdfBuffer,
|
||||
runWhisperAlignmentFromAudioBuffer,
|
||||
} from '@openreader/compute-core/local-runtime';
|
||||
import type { PdfLayoutProgress } from '@openreader/compute-core/contracts';
|
||||
|
||||
async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
|
||||
let timer: NodeJS.Timeout | null = null;
|
||||
try {
|
||||
return await Promise.race([
|
||||
promise,
|
||||
new Promise<T>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error(`${label} timed out after ${timeoutMs}ms`)), timeoutMs);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
async function withIdleTimeoutAndHardCap<T>(input: {
|
||||
run: (touchProgress: () => void) => Promise<T>;
|
||||
idleTimeoutMs: number;
|
||||
hardCapMs: number;
|
||||
label: string;
|
||||
}): Promise<T> {
|
||||
let idleTimer: NodeJS.Timeout | null = null;
|
||||
let hardCapTimer: NodeJS.Timeout | null = null;
|
||||
let settled = false;
|
||||
let rejectTimeout!: (reason: unknown) => void;
|
||||
|
||||
const clearTimers = () => {
|
||||
if (idleTimer) {
|
||||
clearTimeout(idleTimer);
|
||||
idleTimer = null;
|
||||
}
|
||||
if (hardCapTimer) {
|
||||
clearTimeout(hardCapTimer);
|
||||
hardCapTimer = null;
|
||||
}
|
||||
};
|
||||
|
||||
const failTimeout = (kind: 'idle' | 'hard cap', timeoutMs: number) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimers();
|
||||
rejectTimeout(new Error(`${input.label} ${kind} timed out after ${timeoutMs}ms`));
|
||||
};
|
||||
|
||||
const touchProgress = () => {
|
||||
if (settled) return;
|
||||
if (idleTimer) clearTimeout(idleTimer);
|
||||
idleTimer = setTimeout(() => failTimeout('idle', input.idleTimeoutMs), input.idleTimeoutMs);
|
||||
};
|
||||
|
||||
const timeoutPromise = new Promise<never>((_, reject) => {
|
||||
rejectTimeout = reject;
|
||||
hardCapTimer = setTimeout(() => failTimeout('hard cap', input.hardCapMs), input.hardCapMs);
|
||||
touchProgress();
|
||||
});
|
||||
|
||||
try {
|
||||
const result = await Promise.race([input.run(touchProgress), timeoutPromise]);
|
||||
settled = true;
|
||||
clearTimers();
|
||||
return result as T;
|
||||
} catch (error) {
|
||||
settled = true;
|
||||
clearTimers();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export class LocalComputeBackend implements ComputeBackend {
|
||||
readonly mode = 'local' as const;
|
||||
private readonly timeoutConfig = getComputeTimeoutConfig();
|
||||
|
|
|
|||
Loading…
Reference in a new issue