fix(cdp): record the failed run row for jobs read from kafka

nodejs/src/cdp/services/job-queue/job-queue-kafka.test.ts

@@ -1,6 +1,29 @@
-import { migrateKafkaCyclotronInvocation } from './job-queue-kafka'
+import { DateTime } from 'luxon'
+
+import { deserializeInvocation, migrateKafkaCyclotronInvocation, serializeInvocation } from './job-queue-kafka'
 
 describe('CyclotronJobQueueKafka', () => {
+    describe('deserializeInvocation', () => {
+        it.each([
+            ['set', DateTime.utc(2025, 1, 1, 12, 30)],
+            ['unset', undefined],
+        ])('round-trips queueScheduledAt when it is %s', (_, scheduledAt) => {
+            const json = JSON.stringify(
+                serializeInvocation({
+                    id: 'invocation-id',
+                    teamId: 1,
+                    functionId: 'function-id',
+                    state: null,
+                    queue: 'hog',
+                    queuePriority: 0,
+                    queueScheduledAt: scheduledAt,
+                })
+            )
+
+            expect(deserializeInvocation(json).queueScheduledAt?.toISO()).toEqual(scheduledAt?.toISO())
+        })
+    })
+
     describe('migrateKafkaCyclotronInvocation', () => {
         // Pulled from a real job in kafka
         const legacyFormat = {

nodejs/src/cdp/services/job-queue/job-queue-kafka.ts

@@ -3,6 +3,7 @@
  * To make this easier this class is designed to abstract the queue as much as possible from
  * the underlying implementation.
  */
+import { DateTime } from 'luxon'
 import { Message } from 'node-rdkafka'
 import { compress, uncompress } from 'snappy'
 
@@ -184,9 +185,7 @@ export class CyclotronJobQueueKafka implements JobQueue {
 
             // Try to decompress, otherwise just use the value as is
             const decompressedValue = await uncompress(rawValue).catch(() => rawValue)
-            const invocation: CyclotronJobInvocation = migrateKafkaCyclotronInvocation(
-                parseJSON(decompressedValue.toString())
-            )
+            const invocation = deserializeInvocation(decompressedValue.toString())
 
             invocation.queueSource = 'kafka' // NOTE: We always set this here, as we know it came from kafka
             invocations.push(invocation)
@@ -248,3 +247,13 @@ export function serializeInvocation(invocation: CyclotronJobInvocation): Cyclotr
         queueSource: invocation.queueSource,
     }
 }
+
+export function deserializeInvocation(json: string): CyclotronJobInvocation {
+    const invocation = migrateKafkaCyclotronInvocation(parseJSON(json))
+    // JSON.stringify writes the DateTime as an ISO string, and readers call DateTime methods on it.
+    const scheduledAt: unknown = invocation.queueScheduledAt
+    if (typeof scheduledAt === 'string') {
+        invocation.queueScheduledAt = DateTime.fromISO(scheduledAt, { zone: 'utc' })
+    }
+    return invocation
+}