Cron & Serialization Fixes (#1282)

- Fixed inaccurate cron decoding
- Fix an issue where message serialization was ignored during replay
- Fix an issue where `getResult` mishandled
`MAX_RECOVERY_ATTEMPTS_EXCEEDED`

Author: kraftp

src/client.ts

@@ -52,6 +52,7 @@ import {
 import { DBOSExecutor } from './dbos-executor';
 import {
   DBOSAwaitedWorkflowCancelledError,
+  DBOSAwaitedWorkflowExceededMaxRecoveryAttempts,
   DBOSInvalidWorkflowTransitionError,
   DBOSQueueDuplicatedError,
 } from './error';
@@ -195,6 +196,9 @@ export class ClientHandle<R> implements WorkflowHandle<R> {
     if (res?.cancelled) {
       throw new DBOSAwaitedWorkflowCancelledError(this.workflowID);
     }
+    if (res?.maxRecoveryAttemptsExceeded) {
+      throw new DBOSAwaitedWorkflowExceededMaxRecoveryAttempts(this.workflowID);
+    }
     return await DBOSExecutor.reviveResultOrError<R>(res!, this.systemDatabase.getSerializer());
   }
 

src/scheduler/crontab.ts

@@ -68,11 +68,26 @@ function convertWeekDayName(weekExpression: string, items: string[]) {
 }
 
 function weekDayNamesConversion(expression: string) {
-  expression = expression.replace('7', '0');
+  // Don't fold 7 (Sunday) into 0 here: pre-expansion it corrupts ranges like 5-7. See convertSundaySeven.
   expression = convertWeekDayName(expression, weekDays);
   return convertWeekDayName(expression, shortWeekDays);
 }
 
+// Fold day-of-week 7 (Sunday) into 0 after range/step expansion, removing resulting duplicates.
+function convertSundaySeven(expressions: string[]) {
+  const seen = new Set<string>();
+  const folded: string[] = [];
+  for (const value of expressions[5].split(',')) {
+    const day = value === '7' ? '0' : value;
+    if (!seen.has(day)) {
+      seen.add(day);
+      folded.push(day);
+    }
+  }
+  expressions[5] = folded.join(',');
+  return expressions;
+}
+
 function convertAsterisk(expression: string, replacement: string) {
   if (expression.indexOf('*') !== -1) {
     return expression.replace('*', replacement);
@@ -268,6 +283,7 @@ export function convertExpression(crontab: string) {
   expressions = convertSteps(expressions);
 
   expressions = normalizeIntegers(expressions);
+  expressions = convertSundaySeven(expressions);
 
   return expressions.join(' ');
 }

src/system_database.ts

@@ -48,7 +48,7 @@ export interface SystemDatabaseStoredResult {
   maxRecoveryAttemptsExceeded?: boolean;
   childWorkflowID?: string | null;
   functionName?: string;
-  serialization?: string | null; // Only for WF result, not step
+  serialization?: string | null; // For WF result, and for steps that persist a format (recv/getEvent)
 }
 
 /* Exported workflow format for import/export */
@@ -2099,7 +2099,7 @@ export class SystemDatabase {
           `INSERT INTO "${this.schemaName}".workflow_events (workflow_uuid, key, value, serialization)
              VALUES ($1, $2, $3, $4)
              ON CONFLICT (workflow_uuid, key)
-             DO UPDATE SET value = $3
+             DO UPDATE SET value = $3, serialization = $4
              RETURNING workflow_uuid;`,
           [workflowID, key, message, serialization],
         );
@@ -2108,7 +2108,7 @@ export class SystemDatabase {
           `INSERT INTO "${this.schemaName}".workflow_events_history (workflow_uuid, function_id, key, value, serialization)
              VALUES ($1, $2, $3, $4, $5)
              ON CONFLICT (workflow_uuid, function_id, key)
-             DO UPDATE SET value = $4;`,
+             DO UPDATE SET value = $4, serialization = $5;`,
           [workflowID, functionID, key, message, serialization],
         );
         return undefined;
@@ -2151,7 +2151,7 @@ export class SystemDatabase {
             res.functionName!,
           );
         }
-        return { serializedValue: res.output!, serialization: null };
+        return { serializedValue: res.output!, serialization: res.serialization ?? null };
       }
     }
 
@@ -3857,7 +3857,7 @@ export class SystemDatabase {
     await this.#checkIfCanceled(client, workflowID);
 
     const { rows } = await client.query<operation_outputs>(
-      `SELECT output, error, child_workflow_id, function_name
+      `SELECT output, error, child_workflow_id, function_name, serialization
        FROM "${this.schemaName}".operation_outputs
       WHERE workflow_uuid=$1 AND function_id=$2`,
       [workflowID, functionID],
@@ -3870,6 +3870,8 @@ export class SystemDatabase {
         error: rows[0].error,
         childWorkflowID: rows[0].child_workflow_id,
         functionName: rows[0].function_name,
+        // Return serialization so recv/getEvent replay deserializes with the stored format, not the default.
+        serialization: rows[0].serialization,
       };
     }
   }

tests/client.test.ts

@@ -1,10 +1,19 @@
 import { workflow_status } from '../schemas/system_db_schema';
 import { DBOS, DBOSClient, StatusString } from '../src';
 import { globalParams, sleepms } from '../src/utils';
-import { generateDBOSTestConfig, recoverPendingWorkflows, setUpDBOSTestSysDb } from './helpers';
+import {
+  generateDBOSTestConfig,
+  recoverPendingWorkflows,
+  setUpDBOSTestSysDb,
+  setWfAndChildrenToPending,
+} from './helpers';
 import { Client, PoolConfig } from 'pg';
 import { spawnSync } from 'child_process';
-import { DBOSQueueDuplicatedError, DBOSAwaitedWorkflowCancelledError } from '../src/error';
+import {
+  DBOSQueueDuplicatedError,
+  DBOSAwaitedWorkflowCancelledError,
+  DBOSAwaitedWorkflowExceededMaxRecoveryAttempts,
+} from '../src/error';
 import { randomUUID } from 'crypto';
 import { DBOSConfig } from '../src/dbos-executor';
 import { DEFAULT_POOL_SIZE } from '../src/system_database';
@@ -91,6 +100,16 @@ class ClientTest {
   }
 }
 
+const DLQ_MAX_RECOVERY_ATTEMPTS = 2;
+
+class ClientDLQTest {
+  // Succeeds when run, but is forced back to PENDING repeatedly until it exhausts recovery attempts (DLQ).
+  @DBOS.workflow({ maxRecoveryAttempts: DLQ_MAX_RECOVERY_ATTEMPTS })
+  static async dlqWorkflow(): Promise<string> {
+    return Promise.resolve('should-not-be-returned');
+  }
+}
+
 type EnqueueTest = typeof ClientTest.enqueueTest;
 
 function runClientSendWorker(workflowID: string, topic: string, appVersion: string) {
@@ -394,6 +413,31 @@ describe('DBOSClient', () => {
     }
   });
 
+  test('DBOSClient-getResult-throws-on-max-recovery-attempts-exceeded', async () => {
+    // A workflow past its max recovery attempts must make client getResult throw, not silently return null.
+    await DBOS.launch();
+    const client = await DBOSClient.create({ systemDatabaseUrl });
+    try {
+      const handle = await DBOS.startWorkflow(ClientDLQTest).dlqWorkflow();
+      await handle.getResult();
+
+      // Force back to PENDING and recover repeatedly until it exhausts recovery attempts and lands in the DLQ.
+      for (let i = 0; i < DLQ_MAX_RECOVERY_ATTEMPTS; i++) {
+        await setWfAndChildrenToPending(handle.workflowID, false);
+        await (await recoverPendingWorkflows())[0].getResult();
+      }
+      await setWfAndChildrenToPending(handle.workflowID, false);
+      await recoverPendingWorkflows();
+      expect((await handle.getStatus())?.status).toBe(StatusString.MAX_RECOVERY_ATTEMPTS_EXCEEDED);
+
+      await expect(client.retrieveWorkflow(handle.workflowID).getResult()).rejects.toThrow(
+        DBOSAwaitedWorkflowExceededMaxRecoveryAttempts,
+      );
+    } finally {
+      await client.destroy();
+    }
+  });
+
   test('DBOSClient-enqueue-and-get-result-portable', async () => {
     await DBOS.launch();
     await registerTestQueue();

tests/crontab.test.ts

@@ -118,6 +118,15 @@ describe('week-day-names-conversion', () => {
     const weekDays = conversion('* * * * 7').split(' ')[5];
     expect(weekDays).toBe('0');
   });
+
+  // Regression: day-of-week 7 (Sunday) must fold to 0 only after ranges expand, so ranges ending in 7 survive.
+  it('should expand day-of-week ranges that include 7 (Sunday) correctly', () => {
+    expect(conversion('* * * * 5-7').split(' ')[5]).toBe('5,6,0'); // Fri,Sat,Sun
+    expect(conversion('* * * * 6-7').split(' ')[5]).toBe('6,0'); // Sat,Sun
+    expect(conversion('* * * * 1-7').split(' ')[5]).toBe('1,2,3,4,5,6,0'); // every day
+    expect(conversion('* * * * 0-7').split(' ')[5]).toBe('0,1,2,3,4,5,6'); // every day, deduped
+    expect(conversion('* * * * 1,7').split(' ')[5]).toBe('1,0'); // Mon,Sun
+  });
 });
 
 /////////////

tests/portableser.test.ts

@@ -133,6 +133,37 @@ const dateWorkflow = DBOS.registerWorkflow(
   },
 );
 
+// Re-sets one event key twice with different formats; the second setEvent must overwrite value AND serialization.
+const reSetEventWorkflow = DBOS.registerWorkflow(
+  async () => {
+    await DBOS.setEvent('rekey', 10n, { serializationType: 'portable' });
+    await DBOS.setEvent('rekey', 10n, { serializationType: 'native' });
+  },
+  { name: 'reSetEventWorkflow' },
+);
+
+// Workflows for the OAOO replay test: recv/getEvent must keep their recorded serialization across replay.
+const portableEventSetterWF = DBOS.registerWorkflow(
+  async (input: string) => {
+    await DBOS.setEvent('evt', input, { serializationType: 'portable' });
+  },
+  { name: 'portableEventSetterWF' },
+);
+
+const eventGetterWF = DBOS.registerWorkflow(
+  async (setterId: string) => {
+    return await DBOS.getEvent<string>(setterId, 'evt');
+  },
+  { name: 'eventGetterWF' },
+);
+
+const portableRecvWF = DBOS.registerWorkflow(
+  async () => {
+    return await DBOS.recv<string>('topic');
+  },
+  { name: 'portableRecvWF' },
+);
+
 describe('portable-serizlization-tests', () => {
   let config: DBOSConfig;
   let systemDBClient: Client;
@@ -333,6 +364,20 @@ describe('portable-serizlization-tests', () => {
     }
   });
 
+  test('test-setevent-reset-updates-serialization', async () => {
+    // Re-set portable then native: the native format must win so getEvent recovers 10n, not the superjson envelope.
+    const handle = await DBOS.startWorkflow(reSetEventWorkflow)();
+    await handle.getResult();
+
+    const row = await systemDBClient.query<workflow_events>(
+      `SELECT * FROM dbos.workflow_events WHERE workflow_uuid = $1 AND key = $2`,
+      [handle.workflowID, 'rekey'],
+    );
+    expect(row.rows[0].serialization).toBe(DBOSJSON.name());
+
+    expect(await DBOS.getEvent(handle.workflowID, 'rekey')).toBe(10n);
+  });
+
   // Reproduces https://github.com/dbos-inc/dbos-transact-ts/issues/1208
   test('test-portable-void-workflow', async () => {
     // A portable workflow that returns undefined should succeed, not throw
@@ -958,6 +1003,44 @@ describe('custom-serializer-restart-tests', () => {
 
     process.env.DBOS__APPVERSION = undefined;
   });
+
+  test('test-recv-getevent-preserve-serialization-on-replay', async () => {
+    // Under the base64 global serializer, replayed recv/getEvent must use the stored portable format or deserialization throws.
+    process.env.DBOS__APPVERSION = 'v0';
+    await setUpDBOSTestSysDb(config);
+    config.serializer = base64Serializer;
+    DBOS.setConfig(config);
+    await DBOS.launch();
+
+    systemDBClient = new Client({ connectionString: config.systemDatabaseUrl });
+    await systemDBClient.connect();
+
+    // --- getEvent path ---
+    const setter = await DBOS.startWorkflow(portableEventSetterWF)('event-payload');
+    await setter.getResult();
+    // Sanity check: the event was stored in portable format, not base64.
+    const evtRow = await systemDBClient.query<workflow_events>(
+      `SELECT * FROM dbos.workflow_events WHERE workflow_uuid = $1 AND key = $2`,
+      [setter.workflowID, 'evt'],
+    );
+    expect(evtRow.rows[0].serialization).toBe(DBOSPortableJSON.name());
+
+    const getter = await DBOS.startWorkflow(eventGetterWF)(setter.workflowID);
+    expect(await getter.getResult()).toBe('event-payload');
+    // Replay the getter: getEvent hits OAOO and must still return the value.
+    const reGetter = await reexecuteWorkflowById(getter.workflowID);
+    expect(await reGetter?.getResult()).toBe('event-payload');
+
+    // --- recv path ---
+    const receiver = await DBOS.startWorkflow(portableRecvWF)();
+    await DBOS.send(receiver.workflowID, 'msg-payload', 'topic', undefined, { serializationType: 'portable' });
+    expect(await receiver.getResult()).toBe('msg-payload');
+    // Replay the receiver: recv hits OAOO and must still return the message.
+    const reReceiver = await reexecuteWorkflowById(receiver.workflowID);
+    expect(await reReceiver?.getResult()).toBe('msg-payload');
+
+    process.env.DBOS__APPVERSION = undefined;
+  });
 });
 
 // Workflows for async-serializer tests (registered at module level)