From bab3416a36de4cdb53aa80d247a8890a0eedcd0e Mon Sep 17 00:00:00 2001 From: Richard R Date: Tue, 19 May 2026 15:55:35 -0600 Subject: [PATCH] refactor(compute): add timing metrics to worker job results and status Introduce WorkerJobTiming interface to capture queue wait, S3 fetch, and compute durations for compute worker jobs. Update job result and status response types to include timing data. Enhance worker implementation to record and attach timing metrics for alignment jobs, improving observability of job processing latency. Update exports and type usage to support new timing fields. --- compute/core/src/index.ts | 9 ++ compute/worker/src/server.ts | 136 +++++++++++++++++++--- compute/worker/tsconfig.json | 3 +- src/lib/server/compute/worker-contract.ts | 1 + 4 files changed, 130 insertions(+), 19 deletions(-) diff --git a/compute/core/src/index.ts b/compute/core/src/index.ts index 6f9ca15..24afe87 100644 --- a/compute/core/src/index.ts +++ b/compute/core/src/index.ts @@ -31,6 +31,7 @@ export interface WhisperAlignJobRequest { export interface WhisperAlignJobResult { alignments: TTSSentenceAlignment[]; + timing?: WorkerJobTiming; } export interface PdfLayoutJobRequest { @@ -41,6 +42,7 @@ export interface PdfLayoutJobRequest { export interface PdfLayoutJobResult { parsed: ParsedPdfDocument; + timing?: WorkerJobTiming; } export type WorkerJobState = 'queued' | 'running' | 'succeeded' | 'failed'; @@ -50,10 +52,17 @@ export interface WorkerJobErrorShape { code?: string; } +export interface WorkerJobTiming { + queueWaitMs?: number; + s3FetchMs?: number; + computeMs?: number; +} + export interface WorkerJobStatusResponse { status: WorkerJobState; result?: Result; error?: WorkerJobErrorShape; + timing?: WorkerJobTiming; } export async function ensureComputeModels(): Promise { diff --git a/compute/worker/src/server.ts b/compute/worker/src/server.ts index 8226dc3..ebbfcf8 100644 --- a/compute/worker/src/server.ts +++ b/compute/worker/src/server.ts @@ -13,6 +13,7 @@ import { type WhisperAlignJobRequest, type WhisperAlignJobResult, type WorkerJobStatusResponse, + type WorkerJobTiming, } from '@openreader/compute-core'; import { GetObjectCommand, S3Client } from '@aws-sdk/client-s3'; @@ -152,23 +153,68 @@ const layoutSchema = z.object({ }); function mapJobState(job: Job): WorkerJobStatusResponse { + const result = job.returnvalue as Result | undefined; + const resultTiming = readResultTiming(result); + const timing = buildJobTimingSnapshot(job, resultTiming); + if (job.failedReason) { return { status: 'failed', error: { message: job.failedReason || 'Worker job failed', }, + ...(timing ? { timing } : {}), }; } if (typeof job.returnvalue !== 'undefined' && job.finishedOn) { return { status: 'succeeded', result: job.returnvalue as Result, + ...(timing ? { timing } : {}), }; } - if (job.processedOn) return { status: 'running' }; - return { status: 'queued' }; + if (job.processedOn) { + return { + status: 'running', + ...(timing ? { timing } : {}), + }; + } + return { + status: 'queued', + ...(timing ? { timing } : {}), + }; +} + +function readResultTiming(result: Result | undefined): WorkerJobTiming | undefined { + if (!result || typeof result !== 'object') return undefined; + const maybe = result as { timing?: WorkerJobTiming }; + if (!maybe.timing || typeof maybe.timing !== 'object') return undefined; + return maybe.timing; +} + +function toSafeDurationMs(start: number | undefined, end: number | undefined): number | undefined { + if (!Number.isFinite(start) || !Number.isFinite(end)) return undefined; + return Math.max(0, Math.floor((end as number) - (start as number))); +} + +function buildJobTimingSnapshot(job: Job, base?: WorkerJobTiming): WorkerJobTiming | undefined { + const now = Date.now(); + const timestamp = typeof job.timestamp === 'number' ? job.timestamp : undefined; + const processedOn = typeof job.processedOn === 'number' ? job.processedOn : undefined; + + const timing: WorkerJobTiming = { + ...(base ?? {}), + }; + + if (typeof timing.queueWaitMs !== 'number') { + timing.queueWaitMs = processedOn + ? toSafeDurationMs(timestamp, processedOn) + : toSafeDurationMs(timestamp, now); + } + + const hasAnyTiming = Object.values(timing).some((value) => typeof value === 'number' && Number.isFinite(value)); + return hasAnyTiming ? timing : undefined; } async function getQueueDepth(queue: Queue): Promise { @@ -246,9 +292,16 @@ async function main(): Promise { const alignWorker = new Worker( ALIGN_QUEUE_NAME, async (job) => { + const processingStartedAt = Date.now(); + const queueWaitMs = typeof job.timestamp === 'number' + ? Math.max(0, processingStartedAt - job.timestamp) + : undefined; const parsed = alignSchema.parse(job.data); + const s3FetchStartedAt = Date.now(); const audioBuffer = await readObjectByKey(parsed.audioObjectKey); - return withTimeout( + const s3FetchMs = Date.now() - s3FetchStartedAt; + const computeStartedAt = Date.now(); + const result = await withTimeout( runWhisperAlignmentFromAudioBuffer({ audioBuffer, text: parsed.text, @@ -258,6 +311,16 @@ async function main(): Promise { whisperTimeoutMs, 'whisper alignment job', ); + const computeMs = Date.now() - computeStartedAt; + return { + ...result, + timing: { + ...(result.timing ?? {}), + queueWaitMs, + s3FetchMs, + computeMs, + }, + }; }, { connection: redis, @@ -268,9 +331,16 @@ async function main(): Promise { const layoutWorker = new Worker( PDF_LAYOUT_QUEUE_NAME, async (job) => { + const processingStartedAt = Date.now(); + const queueWaitMs = typeof job.timestamp === 'number' + ? Math.max(0, processingStartedAt - job.timestamp) + : undefined; const parsed = layoutSchema.parse(job.data); + const s3FetchStartedAt = Date.now(); const pdfBytes = await readObjectByKey(parsed.documentObjectKey); - return withTimeout( + const s3FetchMs = Date.now() - s3FetchStartedAt; + const computeStartedAt = Date.now(); + const result = await withTimeout( runPdfLayoutFromPdfBuffer({ documentId: parsed.documentId, pdfBytes, @@ -278,6 +348,16 @@ async function main(): Promise { pdfTimeoutMs, 'pdf layout job', ); + const computeMs = Date.now() - computeStartedAt; + return { + ...result, + timing: { + ...(result.timing ?? {}), + queueWaitMs, + s3FetchMs, + computeMs, + }, + }; }, { connection: redis, @@ -285,20 +365,6 @@ async function main(): Promise { }, ); - alignWorker.on('failed', (job, err) => { - console.error('[compute-worker] align job failed', { - jobId: job?.id, - error: err.message, - }); - }); - - layoutWorker.on('failed', (job, err) => { - console.error('[compute-worker] layout job failed', { - jobId: job?.id, - error: err.message, - }); - }); - if (prewarmModels) { await ensureComputeModels(); } @@ -307,6 +373,40 @@ async function main(): Promise { logger: buildLoggerConfig(), }); + alignWorker.on('completed', (job, result) => { + app.log.info({ + queue: ALIGN_QUEUE_NAME, + jobId: job.id, + timing: buildJobTimingSnapshot(job, readResultTiming(result)), + }, 'whisper align job completed'); + }); + + alignWorker.on('failed', (job, err) => { + app.log.error({ + queue: ALIGN_QUEUE_NAME, + jobId: job?.id, + error: err.message, + timing: job ? buildJobTimingSnapshot(job) : undefined, + }, 'whisper align job failed'); + }); + + layoutWorker.on('completed', (job, result) => { + app.log.info({ + queue: PDF_LAYOUT_QUEUE_NAME, + jobId: job.id, + timing: buildJobTimingSnapshot(job, readResultTiming(result)), + }, 'pdf layout job completed'); + }); + + layoutWorker.on('failed', (job, err) => { + app.log.error({ + queue: PDF_LAYOUT_QUEUE_NAME, + jobId: job?.id, + error: err.message, + timing: job ? buildJobTimingSnapshot(job) : undefined, + }, 'pdf layout job failed'); + }); + app.addHook('onRequest', async (request, reply) => { const path = request.url.split('?')[0] ?? request.url; if (path === '/health/live' || path === '/health/ready') return; diff --git a/compute/worker/tsconfig.json b/compute/worker/tsconfig.json index 3e68ae5..5912d1a 100644 --- a/compute/worker/tsconfig.json +++ b/compute/worker/tsconfig.json @@ -3,5 +3,6 @@ "compilerOptions": { "noEmit": true }, - "include": ["src/**/*.ts"] + "include": ["src/**/*.ts"], + "exclude": [] } diff --git a/src/lib/server/compute/worker-contract.ts b/src/lib/server/compute/worker-contract.ts index ec4df20..b43cae9 100644 --- a/src/lib/server/compute/worker-contract.ts +++ b/src/lib/server/compute/worker-contract.ts @@ -7,5 +7,6 @@ export { type WhisperAlignJobResult, type WorkerJobErrorShape, type WorkerJobState, + type WorkerJobTiming, type WorkerJobStatusResponse, } from '@openreader/compute-core';