fix(queue): preserve dtoClass through failed jobs and requeue

config/Migrations/20260922120000_AddDtoClassToFailedJobs.php

@@ -0,0 +1,25 @@
+<?php
+declare(strict_types=1);
+
+use Migrations\BaseMigration;
+
+class AddDtoClassToFailedJobs extends BaseMigration
+{
+    /**
+     * Add nullable DTO class metadata so failed jobs keep enough information
+     * to rehydrate DTOs on requeue. Nullable for backward compatibility with
+     * existing rows and legacy array messages without a DTO.
+     *
+     * @return void
+     */
+    public function change(): void
+    {
+        $table = $this->table('queue_failed_jobs');
+        $table->addColumn('dto_class', 'string', [
+                'length' => 255,
+                'null' => true,
+                'default' => null,
+            ])
+            ->update();
+    }
+}

src/Command/RequeueCommand.php

@@ -138,6 +138,7 @@ public function execute(Arguments $args, ConsoleIo $io): int
                         'config' => $failedJob->config,
                         'priority' => $failedJob->priority,
                         'queue' => $failedJob->queue,
+                        'dtoClass' => $failedJob->dto_class ?? null,
                     ],
                 );
 

src/Command/WorkerCommand.php

@@ -159,7 +159,7 @@ protected function getQueueExtension(Arguments $args, LoggerInterface $logger):
     protected function getLogger(Arguments $args): LoggerInterface
     {
         $logger = null;
-        if (!empty($args->getOption('verbose'))) {
+        if (!(in_array($args->getOption('verbose'), ['', '0'], true) || $args->getOption('verbose') === false || $args->getOption('verbose') === null)) {
             $logger = Log::engine((string)$args->getOption('logger'));
         }
 

src/Listener/FailedJobsListener.php

@@ -67,6 +67,7 @@ public function storeFailedJob(object $event): void
             'class' => $class,
             'method' => $method,
             'data' => json_encode($data),
+            'dto_class' => $originalMessageBody['dtoClass'] ?? null,
             'config' => $requeueOptions['config'],
             'priority' => $requeueOptions['priority'],
             'queue' => $requeueOptions['queue'],

src/Model/Entity/FailedJob.php

@@ -12,6 +12,7 @@
  * @property string $class
  * @property string $method
  * @property string $data
+ * @property string|null $dto_class
  * @property string|null $config
  * @property string|null $priority
  * @property string|null $queue
@@ -37,6 +38,7 @@ class FailedJob extends Entity
         'class' => true,
         'method' => true,
         'data' => true,
+        'dto_class' => true,
         'config' => true,
         'priority' => true,
         'queue' => true,

src/Model/Table/FailedJobsTable.php

@@ -72,6 +72,11 @@ public function validationDefault(Validator $validator): Validator
             ->requirePresence('data', 'create')
             ->notEmptyString('data');
 
+        $validator
+            ->scalar('dto_class')
+            ->maxLength('dto_class', 255)
+            ->allowEmptyString('dto_class');
+
         $validator
             ->scalar('config')
             ->maxLength('config', 255)

tests/Fixture/FailedJobsFixture.php

@@ -27,6 +27,7 @@ public function init(): void
                 'class' => LogToDebugJob::class,
                 'method' => 'execute',
                 'data' => '{"sample_data_1": "sample value", "sample_data_2": 1}',
+                'dto_class' => null,
                 'config' => 'default',
                 'priority' => null,
                 'queue' => 'default',
@@ -38,6 +39,7 @@ public function init(): void
                 'class' => MaxAttemptsIsThreeJob::class,
                 'method' => 'execute',
                 'data' => '{"sample_data_1": "sample value", "sample_data_2": 1}',
+                'dto_class' => null,
                 'config' => 'default',
                 'priority' => null,
                 'queue' => 'default',
@@ -49,6 +51,7 @@ public function init(): void
                 'class' => LogToDebugJob::class,
                 'method' => 'execute',
                 'data' => '{"sample_data_1": "sample value", "sample_data_2": 1}',
+                'dto_class' => null,
                 'config' => 'alternate_config',
                 'priority' => null,
                 'queue' => 'alternate_queue',

tests/TestCase/Command/RequeueCommandTest.php

@@ -19,9 +19,14 @@
 use Cake\Console\TestSuite\ConsoleIntegrationTestTrait;
 use Cake\Core\Configure;
 use Cake\Log\Log;
+use Cake\Queue\Job\Message;
 use Cake\Queue\QueueManager;
 use Cake\Queue\Test\TestCase\QueueTestTrait;
 use Cake\TestSuite\TestCase;
+use Enqueue\Null\NullConnectionFactory;
+use Enqueue\Null\NullMessage;
+use TestApp\Dto\OrderDto;
+use TestApp\Job\DtoJob;
 use TestApp\Job\LogToDebugJob;
 
 /**
@@ -197,4 +202,62 @@ public function testJobsAreRequeuedByConfig()
 
         $this->assertDebugLogContains('Debug job was run');
     }
+
+    public function testRequeuedDtoJobKeepsDtoClass()
+    {
+        $fsQueuePath = TMP . DS . uniqid('queue');
+        QueueManager::setConfig('default', [
+            'url' => 'file:///' . $fsQueuePath,
+            'queue' => 'default',
+        ]);
+
+        /** @var \Cake\Queue\Model\Table\FailedJobsTable $failedJobsTable */
+        $failedJobsTable = $this->getTableLocator()->get('Cake/Queue.FailedJobs');
+        $failedJobsTable->deleteAll(['1=1']);
+
+        $failedJob = $failedJobsTable->newEntity([
+            'class' => DtoJob::class,
+            'method' => 'execute',
+            'data' => json_encode(['id' => 7, 'customer' => 'Acme Corp', 'items' => []]),
+            'dto_class' => OrderDto::class,
+            'config' => 'default',
+            'priority' => null,
+            'queue' => 'default',
+            'exception' => 'boom',
+        ]);
+        $failedJobsTable->saveOrFail($failedJob);
+
+        $this->exec('queue requeue -f');
+
+        $this->assertOutputContains('Requeueing 1 jobs.');
+        $this->assertOutputContains('1 jobs requeued.');
+
+        $fsQueueFile = $fsQueuePath . DS . 'enqueue.app.default';
+        $this->assertFileExists($fsQueueFile);
+
+        $contents = (string)file_get_contents($fsQueueFile);
+        $this->assertStringContainsString('dtoClass', $contents);
+        $this->assertStringContainsString('OrderDto', $contents);
+        $this->assertStringContainsString('Acme Corp', $contents);
+
+        unlink($fsQueueFile);
+    }
+
+    public function testRequeuedDtoJobHydratesAfterRequeue()
+    {
+        $parsedBody = [
+            'class' => [DtoJob::class, 'execute'],
+            'data' => ['id' => 7, 'customer' => 'Acme Corp', 'items' => []],
+            'dtoClass' => OrderDto::class,
+        ];
+        $connectionFactory = new NullConnectionFactory();
+        $context = $connectionFactory->createContext();
+        $message = new Message(new NullMessage((string)json_encode($parsedBody)), $context);
+
+        $dto = $message->getDto(OrderDto::class);
+
+        $this->assertInstanceOf(OrderDto::class, $dto);
+        $this->assertSame(7, $dto->id);
+        $this->assertSame('Acme Corp', $dto->customer);
+    }
 }

tests/TestCase/Listener/FailedJobsListenerTest.php

@@ -33,6 +33,7 @@
 use PHPUnit\Framework\Attributes\DataProvider;
 use RuntimeException;
 use stdClass;
+use TestApp\Dto\OrderDto;
 use TestApp\Job\LogToDebugJob;
 
 class FailedJobsListenerTest extends TestCase
@@ -100,6 +101,84 @@ public function testFailedJobIsAddedWhenEventIsFired()
         $this->assertStringContainsString('some message', $failedJob->exception);
     }
 
+    public function testFailedJobPreservesDtoClass()
+    {
+        $parsedBody = [
+            'class' => [LogToDebugJob::class, 'execute'],
+            'data' => ['id' => 7, 'customer' => 'Acme Corp'],
+            'dtoClass' => OrderDto::class,
+            'requeueOptions' => [
+                'config' => 'example_config',
+                'priority' => 'example_priority',
+                'queue' => 'example_queue',
+            ],
+        ];
+        $messageBody = json_encode($parsedBody);
+        $connectionFactory = new NullConnectionFactory();
+
+        $context = $connectionFactory->createContext();
+        $originalMessage = new NullMessage($messageBody);
+        $message = new Message($originalMessage, $context);
+
+        $event = new Event(
+            'Consumption.LimitAttemptsExtension.failed',
+            $message,
+            ['exception' => 'some message'],
+        );
+
+        /** @var \Cake\Queue\Model\Table\FailedJobsTable $failedJobsTable */
+        $failedJobsTable = $this->getTableLocator()->get('Cake/Queue.FailedJobs');
+        $failedJobsTable->deleteAll(['1=1']);
+
+        EventManager::instance()->on(new FailedJobsListener());
+        EventManager::instance()->dispatch($event);
+
+        $this->assertSame(1, $failedJobsTable->find()->count());
+
+        $failedJob = $failedJobsTable->find()->first();
+
+        $this->assertSame(json_encode(['id' => 7, 'customer' => 'Acme Corp']), $failedJob->data);
+        $this->assertSame(OrderDto::class, $failedJob->dto_class);
+    }
+
+    public function testFailedJobWithoutDtoClassStoresNull()
+    {
+        $parsedBody = [
+            'class' => [LogToDebugJob::class, 'execute'],
+            'data' => ['example_key' => 'example_value'],
+            'requeueOptions' => [
+                'config' => 'example_config',
+                'priority' => 'example_priority',
+                'queue' => 'example_queue',
+            ],
+        ];
+        $messageBody = json_encode($parsedBody);
+        $connectionFactory = new NullConnectionFactory();
+
+        $context = $connectionFactory->createContext();
+        $originalMessage = new NullMessage($messageBody);
+        $message = new Message($originalMessage, $context);
+
+        $event = new Event(
+            'Consumption.LimitAttemptsExtension.failed',
+            $message,
+            ['exception' => 'some message'],
+        );
+
+        /** @var \Cake\Queue\Model\Table\FailedJobsTable $failedJobsTable */
+        $failedJobsTable = $this->getTableLocator()->get('Cake/Queue.FailedJobs');
+        $failedJobsTable->deleteAll(['1=1']);
+
+        EventManager::instance()->on(new FailedJobsListener());
+        EventManager::instance()->dispatch($event);
+
+        $this->assertSame(1, $failedJobsTable->find()->count());
+
+        $failedJob = $failedJobsTable->find()->first();
+
+        $this->assertNull($failedJob->dto_class);
+    }
+
     /**
      * Data provider for testStoreFailedJobException
      *

tests/schema.php

@@ -9,6 +9,7 @@
             'class' => ['type' => 'string', 'length' => 255, 'null' => false, 'default' => null, 'comment' => '', 'precision' => null],
             'method' => ['type' => 'string', 'length' => 255, 'null' => false, 'default' => null, 'comment' => '', 'precision' => null],
             'data' => ['type' => 'text', 'length' => null, 'null' => false, 'default' => null, 'comment' => '', 'precision' => null],
+            'dto_class' => ['type' => 'string', 'length' => 255, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
             'config' => ['type' => 'string', 'length' => 255, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
             'priority' => ['type' => 'string', 'length' => 255, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
             'queue' => ['type' => 'string', 'length' => 255, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],