danvics/openreader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

compute: align local and worker-client timeout behavior

Browse source
This commit is contained in:
Richard R 2026-05-21 16:17:47 -06:00
parent fbb6f484ce
commit e384b66b7d
4 changed files with 120 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
src/app/api/tts/segments/ensure/route.ts
View file

@ -35,6 +35,7 @@ import { getCompute } from '@/lib/server/compute';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
import { resolveTtsModelForProvider } from '@/lib/shared/tts-provider-policy';
import { resolveSegmentAudioUrls } from '@/lib/server/tts/segment-audio-urls';
import { getComputeTimeoutConfig } from '@openreader/compute-core/runtime/timeout-config';
import type {
TTSSegmentInput,
TTSSegmentManifestItem,
@ -45,8 +46,7 @@ import type {
export const runtime = 'nodejs';
export const dynamic = 'force-dynamic';
const GENERATING_STALE_MS = 360_000;
const LOCAL_ALIGNMENT_TIMEOUT_MS = 30_000;
const WORKER_ALIGNMENT_TIMEOUT_MS = 60_000;
const SHARED_ALIGNMENT_TIMEOUT_MS = getComputeTimeoutConfig().whisperTimeoutMs;
function attachDeviceIdCookie(response: NextResponse, deviceId: string | null, didCreate: boolean) {
if (didCreate && deviceId) {
@ -145,10 +145,6 @@ async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: str
}
}
function getAlignmentTimeoutMs(mode: 'local' | 'worker'): number {
return mode === 'worker' ? WORKER_ALIGNMENT_TIMEOUT_MS : LOCAL_ALIGNMENT_TIMEOUT_MS;
}
async function deleteEntryIfUnused(userId: string, segmentEntryId: string): Promise<void> {
const stillReferenced = await db
.select({ segmentId: ttsSegmentVariants.segmentId })
@ -411,7 +407,7 @@ export async function POST(request: NextRequest) {
audioObjectKey: existing.audioKey,
text: segment.text,
}),
getAlignmentTimeoutMs(computeBackend.mode),
SHARED_ALIGNMENT_TIMEOUT_MS,
`Whisper alignment (${computeBackend.mode})`,
)).alignments;
stageTimings.selfHealAlignMs = Date.now() - alignStartedAt;
@ -665,7 +661,7 @@ export async function POST(request: NextRequest) {
audioObjectKey: audioKey,
text: segment.text,
}),
getAlignmentTimeoutMs(computeBackend.mode),
SHARED_ALIGNMENT_TIMEOUT_MS,
`Whisper alignment (${computeBackend.mode})`,
)).alignments;
stageTimings.whisperAlignMs = Date.now() - alignStartedAt;

112
src/lib/server/compute/local.ts
View file

@ -2,13 +2,83 @@ 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 {
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();
async alignWords(input: WhisperAlignInput): Promise<WhisperAlignResult> {
return LOCAL_COMPUTE_LIMITER.run(async () => {
@ -20,12 +90,16 @@ export class LocalComputeBackend implements ComputeBackend {
if (!audioBuffer) {
throw new Error('Local compute alignment requires audioBuffer or audioObjectKey');
}
return runWhisperAlignmentFromAudioBuffer({
audioBuffer,
text: input.text,
cacheKey: input.cacheKey,
lang: input.lang,
});
return withTimeout(
runWhisperAlignmentFromAudioBuffer({
audioBuffer,
text: input.text,
cacheKey: input.cacheKey,
lang: input.lang,
}),
this.timeoutConfig.whisperTimeoutMs,
'local whisper alignment job',
);
});
}
@ -39,15 +113,25 @@ export class LocalComputeBackend implements ComputeBackend {
if (!pdfBytes) {
throw new Error('Local compute PDF layout requires pdfBytes or (documentId + namespace)');
}
return {
parsed: (await runPdfLayoutFromPdfBuffer({
documentId: input.documentId,
pdfBytes,
onPageParsed: (page) => input.onProgress?.({
totalPages: page.totalPages,
pagesParsed: page.pageNumber,
currentPage: page.pageNumber,
phase: 'infer',
parsed: (await withIdleTimeoutAndHardCap({
idleTimeoutMs: Math.max(this.timeoutConfig.pdfTimeoutMs, 1_000),
hardCapMs: this.timeoutConfig.pdfHardCapMs,
label: 'local pdf layout job',
run: async (touchProgress) => runPdfLayoutFromPdfBuffer({
documentId: input.documentId,
pdfBytes,
onPageParsed: (page) => {
touchProgress();
const progress: PdfLayoutProgress = {
totalPages: page.totalPages,
pagesParsed: page.pageNumber,
currentPage: page.pageNumber,
phase: 'infer',
};
return input.onProgress?.(progress);
},
}),
})).parsed,
};

23
src/lib/server/compute/worker.ts
View file

@ -1,5 +1,6 @@
import { createHash, randomUUID } from 'node:crypto';
import type { ComputeBackend, PdfLayoutInput, WhisperAlignInput, WhisperAlignResult } from '@/lib/server/compute/types';
import { getWorkerClientWaitTimeoutMs } from '@openreader/compute-core/runtime/timeout-config';
import type {
PdfLayoutJobRequest,
PdfLayoutJobResult,
@ -20,7 +21,6 @@ class WorkerHttpError extends Error {
}
}
const DEFAULT_WAIT_TIMEOUT_MS = 120_000;
const DEFAULT_RETRIES = 2;
const LOG_PREFIX = '[compute-worker-client]';
const MAX_LOG_DETAIL_CHARS = 600;
@ -236,13 +236,16 @@ export class WorkerComputeBackend implements ComputeBackend {
readonly mode = 'worker' as const;
private readonly baseUrl: string;
private readonly token: string;
private readonly waitTimeoutMs: number;
private readonly waitTimeoutMsByKind: Record<'whisper_align' | 'pdf_layout', number>;
private readonly retries: number;
constructor() {
this.baseUrl = normalizeWorkerBaseUrl(readRequiredEnv('COMPUTE_WORKER_URL'));
this.token = readRequiredEnv('COMPUTE_WORKER_TOKEN');
this.waitTimeoutMs = DEFAULT_WAIT_TIMEOUT_MS;
this.waitTimeoutMsByKind = {
whisper_align: getWorkerClientWaitTimeoutMs('whisper_align'),
pdf_layout: getWorkerClientWaitTimeoutMs('pdf_layout'),
};
this.retries = DEFAULT_RETRIES;
}
@ -268,7 +271,7 @@ export class WorkerComputeBackend implements ComputeBackend {
cacheKey: input.cacheKey ?? null,
lang: input.lang ?? null,
textLength: input.text.length,
waitTimeoutMs: this.waitTimeoutMs,
waitTimeoutMs: this.waitTimeoutMsByKind.whisper_align,
maxRetries: this.retries,
});
@ -292,6 +295,7 @@ export class WorkerComputeBackend implements ComputeBackend {
kind: 'whisper_align',
opKeyHash,
attempt,
waitTimeoutMs: this.waitTimeoutMsByKind.whisper_align,
});
if (final.status !== 'succeeded' || !final.result) {
@ -350,7 +354,7 @@ export class WorkerComputeBackend implements ComputeBackend {
documentId: input.documentId,
namespace: input.namespace ?? null,
documentObjectKey: input.documentObjectKey,
waitTimeoutMs: this.waitTimeoutMs,
waitTimeoutMs: this.waitTimeoutMsByKind.pdf_layout,
maxRetries: this.retries,
});
@ -376,6 +380,7 @@ export class WorkerComputeBackend implements ComputeBackend {
opKeyHash,
documentId: input.documentId,
attempt,
waitTimeoutMs: this.waitTimeoutMsByKind.pdf_layout,
onSnapshot: (snapshot) => {
if (snapshot.progress) {
void input.onProgress?.(snapshot.progress);
@ -489,18 +494,22 @@ export class WorkerComputeBackend implements ComputeBackend {
private async waitForOperation<Result>(
opId: string,
context: Record<string, unknown> & {
waitTimeoutMs?: number;
onSnapshot?: (snapshot: WorkerOperationState<Result>) => void
} = {},
): Promise<WorkerOperationState<Result>> {
const waitTimeoutMs = typeof context.waitTimeoutMs === 'number'
? context.waitTimeoutMs
: this.waitTimeoutMsByKind.whisper_align;
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), this.waitTimeoutMs);
const timeout = setTimeout(() => controller.abort(), waitTimeoutMs);
const startedAt = Date.now();
const traceId = typeof context.traceId === 'string' ? context.traceId : randomUUID();
logWorker('info', 'sse.wait.start', {
...context,
traceId,
opId,
waitTimeoutMs: this.waitTimeoutMs,
waitTimeoutMs,
});
try {

3
tsconfig.json
View file

@ -22,7 +22,8 @@
"@/*": ["./src/*"],
"@openreader/compute-core": ["./compute/core/src/index.ts"],
"@openreader/compute-core/contracts": ["./compute/core/src/contracts.ts"],
"@openreader/compute-core/local-runtime": ["./compute/core/src/local-runtime.ts"]
"@openreader/compute-core/local-runtime": ["./compute/core/src/local-runtime.ts"],
"@openreader/compute-core/runtime/timeout-config": ["./compute/core/src/runtime/timeout-config.ts"]
}
},
"include": ["next-env.d.ts", "**/*.ts", "**/*.tsx", ".next/types/**/*.ts"],