refactor(worker): remove unused job_state kv writes

This commit is contained in:
Richard R 2026-05-26 18:47:53 -06:00
parent e6e029cf8a
commit 0b1ce1c8dc

View file

@ -40,7 +40,6 @@ import type {
WorkerOperationEvent, WorkerOperationEvent,
WhisperAlignJobRequest, WhisperAlignJobRequest,
WhisperAlignJobResult, WhisperAlignJobResult,
WorkerJobErrorShape,
WorkerJobState, WorkerJobState,
WorkerJobTiming, WorkerJobTiming,
WorkerOperationKind, WorkerOperationKind,
@ -95,21 +94,6 @@ interface QueuedJob<TPayload> {
payload: TPayload; payload: TPayload;
} }
interface StoredJobState<Result> {
jobId: string;
opId: string;
opKey: string;
kind: WorkerOperationKind;
status: WorkerJobState;
timestamp: number;
startedAt?: number;
updatedAt: number;
result?: Result;
error?: WorkerJobErrorShape;
timing?: WorkerJobTiming;
progress?: PdfLayoutProgress;
}
interface NatsSession { interface NatsSession {
nc: NatsConnection; nc: NatsConnection;
js: JetStreamClient; js: JetStreamClient;
@ -348,10 +332,6 @@ function isTerminalStatus(status: WorkerJobState): boolean {
return status === 'succeeded' || status === 'failed'; return status === 'succeeded' || status === 'failed';
} }
function jobStateKvKey(jobId: string): string {
return `job_state.${jobId}`;
}
function extractResultRef(kind: WorkerOperationKind, result: unknown): string | undefined { function extractResultRef(kind: WorkerOperationKind, result: unknown): string | undefined {
if (kind !== 'pdf_layout' || !result || typeof result !== 'object') return undefined; if (kind !== 'pdf_layout' || !result || typeof result !== 'object') return undefined;
const maybe = result as { parsedObjectKey?: unknown }; const maybe = result as { parsedObjectKey?: unknown };
@ -665,12 +645,6 @@ async function main(): Promise<void> {
const whisperJobCodec = createJsonCodec<QueuedJob<WhisperAlignJobRequest>>(); const whisperJobCodec = createJsonCodec<QueuedJob<WhisperAlignJobRequest>>();
const layoutJobCodec = createJsonCodec<QueuedJob<PdfLayoutJobRequest>>(); const layoutJobCodec = createJsonCodec<QueuedJob<PdfLayoutJobRequest>>();
const jobStateCodec = createJsonCodec<StoredJobState<WhisperAlignJobResult | PdfLayoutJobResult>>();
const putJobState = async (state: StoredJobState<WhisperAlignJobResult | PdfLayoutJobResult>): Promise<void> => {
const { kv } = await ensureConnected();
await kv.put(jobStateKvKey(state.jobId), jobStateCodec.encode(state));
};
const operationStateStore = new JetStreamOperationStateStore<WhisperAlignJobResult | PdfLayoutJobResult>({ const operationStateStore = new JetStreamOperationStateStore<WhisperAlignJobResult | PdfLayoutJobResult>({
getKv: async () => (await ensureConnected()).kv, getKv: async () => (await ensureConnected()).kv,
@ -686,17 +660,6 @@ async function main(): Promise<void> {
getJs: async () => (await ensureConnected()).js, getJs: async () => (await ensureConnected()).js,
whisperSubject: WHISPER_JOBS_SUBJECT, whisperSubject: WHISPER_JOBS_SUBJECT,
layoutSubject: LAYOUT_JOBS_SUBJECT, layoutSubject: LAYOUT_JOBS_SUBJECT,
onEnqueued: async (job) => {
await putJobState({
jobId: job.jobId,
opId: job.opId,
opKey: job.opKey,
kind: job.kind,
status: 'queued',
timestamp: job.queuedAt,
updatedAt: job.queuedAt,
});
},
}); });
const orchestrator = new OperationOrchestrator({ const orchestrator = new OperationOrchestrator({
@ -1035,18 +998,6 @@ async function main(): Promise<void> {
updatedAt: startedAt, updatedAt: startedAt,
...(queueWaitTiming ? { timing: queueWaitTiming } : {}), ...(queueWaitTiming ? { timing: queueWaitTiming } : {}),
}); });
await putJobState({
jobId: decoded.jobId,
opId: decoded.opId,
opKey: decoded.opKey,
kind: decoded.kind,
status: 'running',
timestamp: decoded.queuedAt,
startedAt,
updatedAt: startedAt,
...(queueWaitTiming ? { timing: queueWaitTiming } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
});
app.log.info({ app.log.info({
worker: input.workerLabel, worker: input.workerLabel,
kind: decoded.kind, kind: decoded.kind,
@ -1072,18 +1023,6 @@ async function main(): Promise<void> {
...(queueWaitTiming ? { timing: queueWaitTiming } : {}), ...(queueWaitTiming ? { timing: queueWaitTiming } : {}),
}); });
} }
await putJobState({
jobId: decoded!.jobId,
opId: decoded!.opId,
opKey: decoded!.opKey,
kind: decoded!.kind,
status: 'running',
timestamp: decoded!.queuedAt,
startedAt,
updatedAt,
...(queueWaitTiming ? { timing: queueWaitTiming } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
});
}; };
heartbeat = setInterval(() => { heartbeat = setInterval(() => {
@ -1094,7 +1033,7 @@ async function main(): Promise<void> {
opId: decoded?.opId, opId: decoded?.opId,
jobId: decoded?.jobId, jobId: decoded?.jobId,
error: toErrorMessage(stateError), error: toErrorMessage(stateError),
}, 'failed to persist running heartbeat state'); }, 'failed to persist operation heartbeat state');
}); });
}, RUNNING_HEARTBEAT_MS); }, RUNNING_HEARTBEAT_MS);
@ -1115,19 +1054,6 @@ async function main(): Promise<void> {
updatedAt: now, updatedAt: now,
...(resultTiming ? { timing: resultTiming } : {}), ...(resultTiming ? { timing: resultTiming } : {}),
}); });
await putJobState({
jobId: decoded.jobId,
opId: decoded.opId,
opKey: decoded.opKey,
kind: decoded.kind,
status: 'succeeded',
timestamp: decoded.queuedAt,
startedAt,
updatedAt: now,
result: result as WhisperAlignJobResult | PdfLayoutJobResult,
...(resultTiming ? { timing: resultTiming } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
});
input.msg.ack(); input.msg.ack();
const terminalDurationMs = safeDurationMs(startedAt, now); const terminalDurationMs = safeDurationMs(startedAt, now);
@ -1164,7 +1090,6 @@ async function main(): Promise<void> {
const queueWaitMs = safeDurationMs(decoded.queuedAt, now); const queueWaitMs = safeDurationMs(decoded.queuedAt, now);
const queueWaitTiming = typeof queueWaitMs === 'number' ? { queueWaitMs } : undefined; const queueWaitTiming = typeof queueWaitMs === 'number' ? { queueWaitMs } : undefined;
const status: WorkerJobState = hasRetriesLeft ? 'running' : 'failed';
const persistOpUpdate = hasRetriesLeft const persistOpUpdate = hasRetriesLeft
? (latestProgress ? (latestProgress
? orchestrator.markProgress({ ? orchestrator.markProgress({
@ -1193,26 +1118,6 @@ async function main(): Promise<void> {
error: toErrorMessage(stateError), error: toErrorMessage(stateError),
}, 'failed to persist operation state'); }, 'failed to persist operation state');
}); });
await putJobState({
jobId: decoded.jobId,
opId: decoded.opId,
opKey: decoded.opKey,
kind: decoded.kind,
status,
timestamp: decoded.queuedAt,
updatedAt: now,
...(status === 'failed' ? { error: { message } } : {}),
...(queueWaitTiming ? { timing: queueWaitTiming } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
}).catch((stateError) => {
app.log.error({
worker: input.workerLabel,
opId: decoded?.opId,
jobId: decoded?.jobId,
error: toErrorMessage(stateError),
}, 'failed to persist job state');
});
} }
if (hasRetriesLeft) { if (hasRetriesLeft) {