No-version workflows are only dequeued by latest-version executors (#1291)

Author: maxdml

src/system_database.ts

@@ -2649,7 +2649,17 @@ export class SystemDatabase {
       }
 
       // Retrieve the first max_tasks workflows in the queue.
-      // Only retrieve workflows of the local version (or without version set)
+      const latestVersionResult = await client.query<{ version_name: string }>(
+        `SELECT version_name
+         FROM "${this.schemaName}".application_versions
+         ORDER BY version_timestamp DESC LIMIT 1`,
+      );
+      const latestVersion = latestVersionResult.rows[0]?.version_name;
+      const isLatestVersion = latestVersion === undefined || latestVersion === appVersion;
+      const versionClause = isLatestVersion
+        ? '(application_version = $3 OR application_version IS NULL)'
+        : 'application_version = $3';
+
       const lockMode = queue.concurrency ? 'FOR UPDATE NOWAIT' : 'FOR UPDATE SKIP LOCKED';
       const limitClause = maxTasks !== Infinity ? `LIMIT ${maxTasks}` : '';
 
@@ -2659,7 +2669,7 @@ export class SystemDatabase {
         FROM "${this.schemaName}".workflow_status
         WHERE status = $1
           AND queue_name = $2
-          AND (application_version IS NULL OR application_version = $3)
+          AND ${versionClause}
           ${partitionFilter.replace('$PARTITION', '$4')}
         ORDER BY priority ASC, created_at ASC
         ${limitClause}

tests/wfqueue.test.ts

@@ -102,6 +102,7 @@ describe('queued-wf-tests-simple', () => {
       TestResumeQueues.queue,
       TestResumeQueuesPartitioned.queue,
       TestConcurrencyAcrossVersions.queue,
+      TestVersionlessDequeue.queue,
       waitFirstQueue,
     ]) {
       await DBOS.registerQueue(ref.name, { onConflict: 'always_update', ...ref.config });
@@ -768,6 +769,18 @@ describe('queued-wf-tests-simple', () => {
     }
   }
 
+  class TestVersionlessDequeue {
+    static queue: QueueRef = {
+      name: 'TestVersionlessDequeue',
+      config: { workerConcurrency: 1, ...testPolling },
+    };
+
+    @DBOS.workflow()
+    static async versionlessWorkflow() {
+      return Promise.resolve(DBOS.workflowID);
+    }
+  }
+
   test('test-concurrency-across-versions', async () => {
     expect(config.systemDatabaseUrl).toBeDefined();
     const client = await DBOSClient.create({ systemDatabaseUrl: config.systemDatabaseUrl! });
@@ -790,6 +803,43 @@ describe('queued-wf-tests-simple', () => {
     await client.destroy();
   });
 
+  test('test-versionless-dequeue-requires-latest-version', async () => {
+    expect(config.systemDatabaseUrl).toBeDefined();
+    const client = await DBOSClient.create({ systemDatabaseUrl: config.systemDatabaseUrl! });
+    const sysdb = DBOSExecutor.globalInstance!.systemDatabase;
+    const workerVersion = globalParams.appVersion;
+
+    // Register a newer application version so this worker is no longer the latest.
+    const newerVersion = `newer-${randomUUID()}`;
+    await sysdb.createApplicationVersion(newerVersion);
+    await sysdb.updateApplicationVersionTimestamp(newerVersion, Date.now() + 1_000_000);
+
+    // Enqueue a version-less workflow (application_version IS NULL): no appVersion field.
+    const versionlessHandle = await client.enqueue({
+      queueName: TestVersionlessDequeue.queue.name,
+      workflowName: 'versionlessWorkflow',
+      workflowClassName: 'TestVersionlessDequeue',
+    });
+
+    // Enqueue a workflow tagged with this worker's own (non-latest) version.
+    const taggedHandle = await DBOS.startWorkflow(TestVersionlessDequeue, {
+      queueName: TestVersionlessDequeue.queue.name,
+    }).versionlessWorkflow();
+
+    // A matching-version worker still dequeues its own version-tagged workflow.
+    await expect(taggedHandle.getResult()).resolves.toBeTruthy();
+
+    // But the version-less workflow stays ENQUEUED: this worker is not the latest version.
+    await sleepms(1000);
+    expect((await versionlessHandle.getStatus())?.status).toBe(StatusString.ENQUEUED);
+
+    // Make this worker the latest version again; the version-less workflow now completes.
+    await sysdb.updateApplicationVersionTimestamp(workerVersion, Date.now() + 2_000_000);
+    await expect(versionlessHandle.getResult()).resolves.toBeTruthy();
+
+    await client.destroy();
+  });
+
   test('test_wait_first_queue', async () => {
     const numTasks = WaitFirstQueueTest.numTasks;
     WaitFirstQueueTest.resetEvents();