From 0b1ce1c8dcf060acb57fe7e942c4ea8ca99bd016 Mon Sep 17 00:00:00 2001
From: Richard R <me@richardr.dev>
Date: Tue, 26 May 2026 18:47:53 -0600
Subject: [PATCH] refactor(worker): remove unused job_state kv writes

---
 compute/worker/src/server.ts | 97 +-----------------------------------
 1 file changed, 1 insertion(+), 96 deletions(-)

diff --git a/compute/worker/src/server.ts b/compute/worker/src/server.ts
index 3df73f5..303ac9b 100644
--- a/compute/worker/src/server.ts
+++ b/compute/worker/src/server.ts
@@ -40,7 +40,6 @@ import type {
   WorkerOperationEvent,
   WhisperAlignJobRequest,
   WhisperAlignJobResult,
-  WorkerJobErrorShape,
   WorkerJobState,
   WorkerJobTiming,
   WorkerOperationKind,
@@ -95,21 +94,6 @@ interface QueuedJob<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 {
   nc: NatsConnection;
   js: JetStreamClient;
@@ -348,10 +332,6 @@ function isTerminalStatus(status: WorkerJobState): boolean {
   return status === 'succeeded' || status === 'failed';
 }
 
-function jobStateKvKey(jobId: string): string {
-  return `job_state.${jobId}`;
-}
-
 function extractResultRef(kind: WorkerOperationKind, result: unknown): string | undefined {
   if (kind !== 'pdf_layout' || !result || typeof result !== 'object') return undefined;
   const maybe = result as { parsedObjectKey?: unknown };
@@ -665,12 +645,6 @@ async function main(): Promise<void> {
 
   const whisperJobCodec = createJsonCodec<QueuedJob<WhisperAlignJobRequest>>();
   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>({
     getKv: async () => (await ensureConnected()).kv,
@@ -686,17 +660,6 @@ async function main(): Promise<void> {
     getJs: async () => (await ensureConnected()).js,
     whisperSubject: WHISPER_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({
@@ -1035,18 +998,6 @@ async function main(): Promise<void> {
         updatedAt: startedAt,
         ...(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({
         worker: input.workerLabel,
         kind: decoded.kind,
@@ -1072,18 +1023,6 @@ async function main(): Promise<void> {
             ...(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(() => {
@@ -1094,7 +1033,7 @@ async function main(): Promise<void> {
             opId: decoded?.opId,
             jobId: decoded?.jobId,
             error: toErrorMessage(stateError),
-          }, 'failed to persist running heartbeat state');
+          }, 'failed to persist operation heartbeat state');
         });
       }, RUNNING_HEARTBEAT_MS);
 
@@ -1115,19 +1054,6 @@ async function main(): Promise<void> {
         updatedAt: now,
         ...(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();
       const terminalDurationMs = safeDurationMs(startedAt, now);
@@ -1164,7 +1090,6 @@ async function main(): Promise<void> {
         const queueWaitMs = safeDurationMs(decoded.queuedAt, now);
         const queueWaitTiming = typeof queueWaitMs === 'number' ? { queueWaitMs } : undefined;
 
-        const status: WorkerJobState = hasRetriesLeft ? 'running' : 'failed';
         const persistOpUpdate = hasRetriesLeft
           ? (latestProgress
             ? orchestrator.markProgress({
@@ -1193,26 +1118,6 @@ async function main(): Promise<void> {
             error: toErrorMessage(stateError),
           }, '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) {