From 8df30d1ea597bd6b95e71377e6a63ba114833bc0 Mon Sep 17 00:00:00 2001 From: Yevgeny Tomenko Date: Tue, 22 Sep 2026 21:48:55 +0300 Subject: [PATCH 1/3] feat(queue): add metadata envelope to queue messages QueueManager::push() accepts a metadata option recorded on the message body as a sibling of data, Message::getMetadata() exposes it, and failed jobs persist it through store and requeue. --- ...20260922163025_AddMetadataToFailedJobs.php | 25 ++++++ src/Command/RequeueCommand.php | 1 + src/Job/Message.php | 15 ++++ src/Listener/FailedJobsListener.php | 3 + src/Model/Entity/FailedJob.php | 20 +++++ src/Model/Table/FailedJobsTable.php | 4 + src/QueueManager.php | 9 +++ tests/Fixture/FailedJobsFixture.php | 3 + tests/TestCase/Command/RequeueCommandTest.php | 40 ++++++++++ tests/TestCase/Job/MessageTest.php | 42 ++++++++++ .../Listener/FailedJobsListenerTest.php | 78 +++++++++++++++++++ tests/TestCase/QueueManagerTest.php | 35 +++++++++ tests/schema.php | 1 + 13 files changed, 276 insertions(+) create mode 100644 config/Migrations/20260922163025_AddMetadataToFailedJobs.php diff --git a/config/Migrations/20260922163025_AddMetadataToFailedJobs.php b/config/Migrations/20260922163025_AddMetadataToFailedJobs.php new file mode 100644 index 0000000..3638b0f --- /dev/null +++ b/config/Migrations/20260922163025_AddMetadataToFailedJobs.php @@ -0,0 +1,25 @@ +table('queue_failed_jobs'); + $table->addColumn('metadata', 'text', [ + 'null' => true, + 'default' => null, + ]) + ->update(); + } +} diff --git a/src/Command/RequeueCommand.php b/src/Command/RequeueCommand.php index ce00f7c..a4122e0 100644 --- a/src/Command/RequeueCommand.php +++ b/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, + 'metadata' => $failedJob->decoded_metadata ?? [], ], ); diff --git a/src/Job/Message.php b/src/Job/Message.php index 521351d..b4fd15f 100644 --- a/src/Job/Message.php +++ b/src/Job/Message.php @@ -204,6 +204,21 @@ public function getDto(string $dtoClass): object return $dto; } + /** + * Get the envelope metadata recorded on the message body at dispatch time. + * + * Metadata carries bookkeeping fields that belong next to the payload, + * not inside it, so `data` stays pure for DTO hydration. + * + * @return array + */ + public function getMetadata(): array + { + $metadata = $this->parsedBody['metadata'] ?? []; + + return is_array($metadata) ? $metadata : []; + } + /** * The maximum number of attempts allowed by the job. */ diff --git a/src/Listener/FailedJobsListener.php b/src/Listener/FailedJobsListener.php index c983fb4..e9e98d8 100644 --- a/src/Listener/FailedJobsListener.php +++ b/src/Listener/FailedJobsListener.php @@ -67,6 +67,9 @@ public function storeFailedJob(object $event): void 'class' => $class, 'method' => $method, 'data' => json_encode($data), + 'metadata' => isset($originalMessageBody['metadata']) + ? (string)json_encode($originalMessageBody['metadata']) + : null, 'config' => $requeueOptions['config'], 'priority' => $requeueOptions['priority'], 'queue' => $requeueOptions['queue'], diff --git a/src/Model/Entity/FailedJob.php b/src/Model/Entity/FailedJob.php index 62f1fc4..164342d 100644 --- a/src/Model/Entity/FailedJob.php +++ b/src/Model/Entity/FailedJob.php @@ -12,6 +12,7 @@ * @property string $class * @property string $method * @property string $data + * @property string|null $metadata * @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, + 'metadata' => true, 'config' => true, 'priority' => true, 'queue' => true, @@ -52,4 +54,22 @@ protected function _getDecodedData(): array { return json_decode($this->data, true); } + + /** + * Envelope metadata as an array. Empty when the job was stored without + * a metadata envelope. + * + * @see \Cake\Queue\Model\Entity\FailedJob::$decoded_metadata + * @return array + */ + protected function _getDecodedMetadata(): array + { + if (empty($this->metadata)) { + return []; + } + + $decoded = json_decode((string)$this->metadata, true); + + return is_array($decoded) ? $decoded : []; + } } diff --git a/src/Model/Table/FailedJobsTable.php b/src/Model/Table/FailedJobsTable.php index 208b325..63d6df5 100644 --- a/src/Model/Table/FailedJobsTable.php +++ b/src/Model/Table/FailedJobsTable.php @@ -72,6 +72,10 @@ public function validationDefault(Validator $validator): Validator ->requirePresence('data', 'create') ->notEmptyString('data'); + $validator + ->scalar('metadata') + ->allowEmptyString('metadata'); + $validator ->scalar('config') ->maxLength('config', 255) diff --git a/src/QueueManager.php b/src/QueueManager.php index ff8fafd..6111c6c 100644 --- a/src/QueueManager.php +++ b/src/QueueManager.php @@ -223,6 +223,10 @@ public static function engine(string $name): SimpleClient * - `expires` - Time (in integer seconds) after which the message expires. * The message will be removed from the queue if this time is exceeded * and it has not been consumed. Default `null`. + * - `metadata` - Optional envelope data recorded on the message body next + * to `data` (e.g. `tags`, `_uniqueId`, `batch_id`). Unlike `data` it is + * never passed to the job or hydrated into a DTO. Omitted from the body + * when empty or not an array. Default `[]`. * - `priority` - Valid values: * - `\Enqueue\Client\MessagePriority::VERY_LOW` * - `\Enqueue\Client\MessagePriority::LOW` @@ -297,6 +301,11 @@ public static function push(string|array $className, array|object $data = [], ar $body['dtoClass'] = $dtoClass; } + $metadata = $options['metadata'] ?? null; + if (is_array($metadata) && $metadata !== []) { + $body['metadata'] = $metadata; + } + $message = new ClientMessage($body); if (isset($options['delay'])) { diff --git a/tests/Fixture/FailedJobsFixture.php b/tests/Fixture/FailedJobsFixture.php index c44743e..09dc47d 100644 --- a/tests/Fixture/FailedJobsFixture.php +++ b/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}', + 'metadata' => 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}', + 'metadata' => 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}', + 'metadata' => null, 'config' => 'alternate_config', 'priority' => null, 'queue' => 'alternate_queue', diff --git a/tests/TestCase/Command/RequeueCommandTest.php b/tests/TestCase/Command/RequeueCommandTest.php index 5887b51..6dc856f 100644 --- a/tests/TestCase/Command/RequeueCommandTest.php +++ b/tests/TestCase/Command/RequeueCommandTest.php @@ -197,4 +197,44 @@ public function testJobsAreRequeuedByConfig() $this->assertDebugLogContains('Debug job was run'); } + + public function testRequeuedJobKeepsMetadata() + { + $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' => LogToDebugJob::class, + 'method' => 'execute', + 'data' => json_encode(['example_key' => 'example_value']), + 'metadata' => json_encode(['tags' => ['finance'], '_uniqueId' => 'abc123']), + '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('metadata', $contents); + $this->assertStringContainsString('finance', $contents); + $this->assertStringContainsString('abc123', $contents); + + unlink($fsQueueFile); + } } diff --git a/tests/TestCase/Job/MessageTest.php b/tests/TestCase/Job/MessageTest.php index d8c384e..3d8a6fc 100644 --- a/tests/TestCase/Job/MessageTest.php +++ b/tests/TestCase/Job/MessageTest.php @@ -248,6 +248,48 @@ public function testGetDtoThrowsForMissingExpectedClass() $message->getDto('TestApp\Dto\DoesNotExist'); } + /** + * Test that envelope metadata is exposed separately from the payload. + * + * @return void + */ + public function testGetMetadata() + { + $parsedBody = [ + 'class' => [WelcomeMailer::class, 'welcome'], + 'data' => ['id' => 7], + 'metadata' => [ + 'tags' => ['finance', 'orders'], + '_uniqueId' => 'abc123', + ], + ]; + $connectionFactory = new NullConnectionFactory(); + $context = $connectionFactory->createContext(); + $originalMessage = new NullMessage((string)json_encode($parsedBody)); + $message = new Message($originalMessage, $context); + + $this->assertSame($parsedBody['metadata'], $message->getMetadata()); + // The payload stays pure: no envelope keys leak into the job data. + $this->assertSame(['id' => 7], $message->getArgument()); + } + + /** + * Test that missing metadata defaults to an empty array. + * + * @return void + */ + public function testGetMetadataDefaultsToEmpty() + { + $connectionFactory = new NullConnectionFactory(); + $context = $connectionFactory->createContext(); + + $plain = new Message(new NullMessage((string)json_encode([ + 'class' => [WelcomeMailer::class, 'welcome'], + 'data' => ['id' => 7], + ])), $context); + $this->assertSame([], $plain->getMetadata()); + } + /** * Test that invalid classes cannot be made into callables. * diff --git a/tests/TestCase/Listener/FailedJobsListenerTest.php b/tests/TestCase/Listener/FailedJobsListenerTest.php index d5a56e9..a1c9bc2 100644 --- a/tests/TestCase/Listener/FailedJobsListenerTest.php +++ b/tests/TestCase/Listener/FailedJobsListenerTest.php @@ -100,6 +100,84 @@ public function testFailedJobIsAddedWhenEventIsFired() $this->assertStringContainsString('some message', $failedJob->exception); } + public function testFailedJobPreservesMetadata() + { + $parsedBody = [ + 'class' => [LogToDebugJob::class, 'execute'], + 'data' => ['example_key' => 'example_value'], + 'metadata' => ['tags' => ['finance'], '_uniqueId' => 'abc123'], + '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($parsedBody['metadata'], $failedJob->decoded_metadata); + } + + public function testFailedJobWithoutMetadataStoresNull() + { + $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->metadata); + $this->assertSame([], $failedJob->decoded_metadata); + } + /** * Data provider for testStoreFailedJobException * diff --git a/tests/TestCase/QueueManagerTest.php b/tests/TestCase/QueueManagerTest.php index 8999eb3..5ff9867 100644 --- a/tests/TestCase/QueueManagerTest.php +++ b/tests/TestCase/QueueManagerTest.php @@ -304,6 +304,41 @@ public function testPushWithoutDtoDoesNotAddDtoClass() $this->assertStringNotContainsString('dtoClass', $contents); } + public function testPushWithMetadata() + { + QueueManager::setConfig('test', [ + 'url' => $this->getFsQueueUrl(), + 'queue' => 'test', + ]); + + QueueManager::push(LogToDebugJob::class, ['id' => 7], [ + 'config' => 'test', + 'metadata' => ['tags' => ['finance'], '_uniqueId' => 'abc123'], + ]); + + $fsQueueFile = $this->getFsQueueUrl() . DS . 'enqueue.app.test'; + $this->assertFileExists($fsQueueFile); + $contents = file_get_contents($fsQueueFile); + $this->assertStringContainsString('metadata', $contents); + $this->assertStringContainsString('finance', $contents); + $this->assertStringContainsString('abc123', $contents); + } + + public function testPushWithoutMetadataDoesNotAddMetadata() + { + QueueManager::setConfig('test', [ + 'url' => $this->getFsQueueUrl(), + 'queue' => 'test', + ]); + + QueueManager::push(LogToDebugJob::class, ['id' => 7], ['config' => 'test']); + + $fsQueueFile = $this->getFsQueueUrl() . DS . 'enqueue.app.test'; + $this->assertFileExists($fsQueueFile); + $contents = file_get_contents($fsQueueFile); + $this->assertStringNotContainsString('metadata', $contents); + } + public function testUniqueMessageIsQueuedOnlyOnce() { QueueManager::setConfig('test', [ diff --git a/tests/schema.php b/tests/schema.php index 3aeba16..1dfba5f 100644 --- a/tests/schema.php +++ b/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], + 'metadata' => ['type' => 'text', 'length' => null, '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], From ef7c1858b786212ce590a3769e26cc1015afddda Mon Sep 17 00:00:00 2001 From: Yevgeny Tomenko Date: Sat, 17 Oct 2026 22:38:15 +0300 Subject: [PATCH 2/3] Store failed-job data and metadata as native json columns Follow-up to the metadata envelope: use the native `json` column type for `data` and `metadata` instead of `text` with manual encode/decode, so the ORM handles serialization via JsonType. --- composer.json | 1 + ...20260922163025_AddMetadataToFailedJobs.php | 10 +++++- docs/en/failed-jobs.md | 10 ++++++ docs/en/jobs.md | 33 +++++++++++++++++++ docs/en/upgrading.md | 8 +++++ src/Command/RequeueCommand.php | 4 +-- src/Command/WorkerCommand.php | 2 +- src/Job/Message.php | 2 +- src/Listener/FailedJobsListener.php | 6 ++-- src/Model/Entity/FailedJob.php | 33 ++----------------- src/Model/Table/FailedJobsTable.php | 16 ++++++--- tests/Fixture/FailedJobsFixture.php | 9 +++-- tests/TestCase/Command/RequeueCommandTest.php | 4 +-- .../Listener/FailedJobsListenerTest.php | 5 ++- tests/schema.php | 4 +-- 15 files changed, 93 insertions(+), 54 deletions(-) diff --git a/composer.json b/composer.json index cbe9d9c..53f147b 100644 --- a/composer.json +++ b/composer.json @@ -29,6 +29,7 @@ "require-dev": { "cakephp/bake": "^3.5.1", "cakephp/cakephp-codesniffer": "^5.3", + "enqueue/enqueue": "^0.10.27", "enqueue/fs": "^0.10", "phpunit/phpunit": "^11.5.3 || ^12.1.3 || ^13.0" }, diff --git a/config/Migrations/20260922163025_AddMetadataToFailedJobs.php b/config/Migrations/20260922163025_AddMetadataToFailedJobs.php index 3638b0f..613ff8e 100644 --- a/config/Migrations/20260922163025_AddMetadataToFailedJobs.php +++ b/config/Migrations/20260922163025_AddMetadataToFailedJobs.php @@ -11,15 +11,23 @@ class AddMetadataToFailedJobs extends BaseMigration * for backward compatibility with existing rows and legacy messages * dispatched without a metadata envelope. * + * Also converts `data` to a native `json` column so the ORM encodes/decodes + * it automatically instead of relying on the manual JSON handling of the + * legacy `text` column. + * * @return void */ public function change(): void { $table = $this->table('queue_failed_jobs'); - $table->addColumn('metadata', 'text', [ + $table->addColumn('metadata', 'json', [ 'null' => true, 'default' => null, ]) + ->changeColumn('data', 'json', [ + 'null' => false, + 'default' => null, + ]) ->update(); } } diff --git a/docs/en/failed-jobs.md b/docs/en/failed-jobs.md index e0038a2..fa95a37 100644 --- a/docs/en/failed-jobs.md +++ b/docs/en/failed-jobs.md @@ -11,6 +11,16 @@ If `storeFailedJobs` is enabled, failed jobs are rejected, written to the `queue Your broker may also support dead-letter queues. Those are separate transport-level features. Queue's failed-job storage is transport-agnostic and only tracks jobs that exhaust Queue's retry handling. +## Metadata Envelope + +The `queue_failed_jobs` table has a nullable `metadata` column that preserves the message envelope (`tags`, `_uniqueId`, `batch_id`, ...) through store and requeue: `queue requeue` pushes `data` and `metadata` back together. Rows stored before the envelope existed have `NULL` metadata and requeue exactly as before. Make sure the plugin migrations are applied: + +```bash +bin/cake migrations migrate --plugin Cake/Queue +``` + +Both `data` and `metadata` are native `json` columns, so the ORM encodes and decodes them automatically — `$failedJob->data` and `$failedJob->metadata` are already arrays, no manual `json_decode()` needed. + ## Requeue Failed Jobs Push failed jobs back onto the queue: diff --git a/docs/en/jobs.md b/docs/en/jobs.md index d6a3666..1a43c09 100644 --- a/docs/en/jobs.md +++ b/docs/en/jobs.md @@ -42,6 +42,7 @@ Job classes can use constructor injection in the same way as controllers or comm The `Message` object exposes: - `getArgument($key = null, $default = null)` to fetch the full payload or a nested value using `Hash::get()` notation. +- `getMetadata()` to read the envelope metadata recorded next to the payload (see "Message Metadata Envelope"). - `getContext()` to access the original queue context. - `getOriginalMessage()` to access the broker-specific message object. - `getParsedBody()` to inspect the parsed queue body. @@ -87,6 +88,7 @@ Supported options: - `config`: queue config name. Defaults to `default`. - `delay`: delay processing by a number of seconds. Broker support varies. - `expires`: expire the message after a number of seconds if it has not been consumed. +- `metadata`: optional envelope array recorded on the message body next to `data` (e.g. `tags`, `_uniqueId`, `batch_id`). Unlike `data` it is never passed to the job or hydrated into a DTO. Omitted from the body when empty. Defaults to `[]`. - `priority`: one of `\Enqueue\Client\MessagePriority::VERY_LOW`, `LOW`, `NORMAL`, `HIGH`, or `VERY_HIGH`. - `queue`: queue name to use. Defaults to the configured queue, then `default`. @@ -155,3 +157,34 @@ Hydration mirrors the DTO conventions used elsewhere in CakePHP (`#[RequestToDto ``` - **`createFromArray()` factory** — if the DTO class defines a static `createFromArray(array $data, bool $nested = false): static` method, it's used instead of reflection. + +## Message Metadata Envelope + +Besides the payload (`data`), a message can carry an envelope of bookkeeping fields in a sibling `metadata` section: + +```php +QueueManager::push(ProcessOrderJob::class, $order, [ + 'metadata' => ['tags' => ['finance'], '_uniqueId' => $uuid], +]); +``` + +```php +public function execute(Message $message): ?string +{ + $order = $message->getDto(OrderDto::class); + $tags = $message->getMetadata()['tags'] ?? []; + + return Processor::ACK; +} +``` + +Rules: + +- `metadata` travels next to `data`, never inside it, so the payload stays pure and DTO-safe. +- It is never passed to the job and never hydrated into a DTO — read it with `Message::getMetadata()`, which returns `[]` for legacy messages without an envelope. +- It is omitted from the message body when empty, so pushes without it produce byte-identical bodies. +- It survives failed-job storage and requeue (see Failed Jobs). + +### Third-party payload + +The envelope is intentionally opaque to this plugin: it is a carrier for third-party producer and monitoring software, which may write bookkeeping fields at dispatch time and read them back on the worker side. The base plugin transports and persists the envelope but never interprets its keys — what a tag means, and which keys matter, is defined by those consumers. diff --git a/docs/en/upgrading.md b/docs/en/upgrading.md index 52e4285..d755e80 100644 --- a/docs/en/upgrading.md +++ b/docs/en/upgrading.md @@ -10,3 +10,11 @@ Queue 3.0 requires PHP 8.2 or later and CakePHP 5.4 or later. ```php $this->addPlugin('Cake/Queue'); ``` + +## Failed-Job `data` Is Now a Native JSON Column + +The `queue_failed_jobs.data` column (and the new `metadata` column, see Failed Jobs) uses the native `json` type, so the ORM returns arrays directly: `$failedJob->data` is an array, not a JSON string. The `FailedJob::decoded_data` and `FailedJob::decoded_metadata` accessors were removed — read `$failedJob->data` and `$failedJob->metadata` instead. Run the plugin migrations to get the new schema: + +```bash +bin/cake migrations migrate --plugin Cake/Queue +``` diff --git a/src/Command/RequeueCommand.php b/src/Command/RequeueCommand.php index a4122e0..f85fa68 100644 --- a/src/Command/RequeueCommand.php +++ b/src/Command/RequeueCommand.php @@ -133,12 +133,12 @@ public function execute(Arguments $args, ConsoleIo $io): int try { QueueManager::push( [$failedJob->class, $failedJob->method], - $failedJob->decoded_data, + $failedJob->data, [ 'config' => $failedJob->config, 'priority' => $failedJob->priority, 'queue' => $failedJob->queue, - 'metadata' => $failedJob->decoded_metadata ?? [], + 'metadata' => $failedJob->metadata ?? [], ], ); diff --git a/src/Command/WorkerCommand.php b/src/Command/WorkerCommand.php index 44605d1..7efa802 100644 --- a/src/Command/WorkerCommand.php +++ b/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')); } diff --git a/src/Job/Message.php b/src/Job/Message.php index b4fd15f..4b328d1 100644 --- a/src/Job/Message.php +++ b/src/Job/Message.php @@ -207,7 +207,7 @@ public function getDto(string $dtoClass): object /** * Get the envelope metadata recorded on the message body at dispatch time. * - * Metadata carries bookkeeping fields that belong next to the payload, + * Metadata carries bookkeeping fields that belong next to the payload, * not inside it, so `data` stays pure for DTO hydration. * * @return array diff --git a/src/Listener/FailedJobsListener.php b/src/Listener/FailedJobsListener.php index e9e98d8..ef358ce 100644 --- a/src/Listener/FailedJobsListener.php +++ b/src/Listener/FailedJobsListener.php @@ -66,10 +66,8 @@ public function storeFailedJob(object $event): void $failedJob = $failedJobsTable->newEntity([ 'class' => $class, 'method' => $method, - 'data' => json_encode($data), - 'metadata' => isset($originalMessageBody['metadata']) - ? (string)json_encode($originalMessageBody['metadata']) - : null, + 'data' => $data, + 'metadata' => $originalMessageBody['metadata'] ?? null, 'config' => $requeueOptions['config'], 'priority' => $requeueOptions['priority'], 'queue' => $requeueOptions['queue'], diff --git a/src/Model/Entity/FailedJob.php b/src/Model/Entity/FailedJob.php index 164342d..3735d23 100644 --- a/src/Model/Entity/FailedJob.php +++ b/src/Model/Entity/FailedJob.php @@ -11,15 +11,13 @@ * @property int $id * @property string $class * @property string $method - * @property string $data - * @property string|null $metadata + * @property array $data + * @property array|null $metadata * @property string|null $config * @property string|null $priority * @property string|null $queue * @property string|null $exception * @property \Cake\I18n\DateTime|null $created - * - * @property array $decoded_data */ class FailedJob extends Entity { @@ -45,31 +43,4 @@ class FailedJob extends Entity 'exception' => true, 'created' => true, ]; - - /** - * @see \Cake\Queue\Model\Entity\FailedJob::$decoded_data - * @return array - */ - protected function _getDecodedData(): array - { - return json_decode($this->data, true); - } - - /** - * Envelope metadata as an array. Empty when the job was stored without - * a metadata envelope. - * - * @see \Cake\Queue\Model\Entity\FailedJob::$decoded_metadata - * @return array - */ - protected function _getDecodedMetadata(): array - { - if (empty($this->metadata)) { - return []; - } - - $decoded = json_decode((string)$this->metadata, true); - - return is_array($decoded) ? $decoded : []; - } } diff --git a/src/Model/Table/FailedJobsTable.php b/src/Model/Table/FailedJobsTable.php index 63d6df5..d157460 100644 --- a/src/Model/Table/FailedJobsTable.php +++ b/src/Model/Table/FailedJobsTable.php @@ -40,6 +40,14 @@ public function initialize(array $config): void $this->setDisplayField('id'); $this->setPrimaryKey('id'); + // Force the JSON type on every driver. SQLite stores json columns as + // text, so introspection alone would not apply Cake\ORM's JsonType. + $schema = $this->getSchema(); + $schema->setColumnType('data', 'json'); + if ($schema->hasColumn('metadata')) { + $schema->setColumnType('metadata', 'json'); + } + $this->addBehavior('Timestamp'); } @@ -68,13 +76,13 @@ public function validationDefault(Validator $validator): Validator ->notEmptyString('method'); $validator - ->scalar('data') ->requirePresence('data', 'create') - ->notEmptyString('data'); + ->allowEmptyArray('data') + ->array('data'); $validator - ->scalar('metadata') - ->allowEmptyString('metadata'); + ->allowEmptyArray('metadata') + ->array('metadata'); $validator ->scalar('config') diff --git a/tests/Fixture/FailedJobsFixture.php b/tests/Fixture/FailedJobsFixture.php index 09dc47d..bfdf393 100644 --- a/tests/Fixture/FailedJobsFixture.php +++ b/tests/Fixture/FailedJobsFixture.php @@ -26,7 +26,7 @@ public function init(): void 'id' => 1, 'class' => LogToDebugJob::class, 'method' => 'execute', - 'data' => '{"sample_data_1": "sample value", "sample_data_2": 1}', + 'data' => ['sample_data_1' => 'sample value', 'sample_data_2' => 1], 'metadata' => null, 'config' => 'default', 'priority' => null, @@ -38,7 +38,7 @@ public function init(): void 'id' => 2, 'class' => MaxAttemptsIsThreeJob::class, 'method' => 'execute', - 'data' => '{"sample_data_1": "sample value", "sample_data_2": 1}', + 'data' => ['sample_data_1' => 'sample value', 'sample_data_2' => 1], 'metadata' => null, 'config' => 'default', 'priority' => null, @@ -50,7 +50,7 @@ public function init(): void 'id' => 3, 'class' => LogToDebugJob::class, 'method' => 'execute', - 'data' => '{"sample_data_1": "sample value", "sample_data_2": 1}', + 'data' => ['sample_data_1' => 'sample value', 'sample_data_2' => 1], 'metadata' => null, 'config' => 'alternate_config', 'priority' => null, @@ -60,5 +60,8 @@ public function init(): void ], ]; parent::init(); + + $this->_schema->setColumnType('data', 'json'); + $this->_schema->setColumnType('metadata', 'json'); } } diff --git a/tests/TestCase/Command/RequeueCommandTest.php b/tests/TestCase/Command/RequeueCommandTest.php index 6dc856f..90dff0c 100644 --- a/tests/TestCase/Command/RequeueCommandTest.php +++ b/tests/TestCase/Command/RequeueCommandTest.php @@ -213,8 +213,8 @@ public function testRequeuedJobKeepsMetadata() $failedJob = $failedJobsTable->newEntity([ 'class' => LogToDebugJob::class, 'method' => 'execute', - 'data' => json_encode(['example_key' => 'example_value']), - 'metadata' => json_encode(['tags' => ['finance'], '_uniqueId' => 'abc123']), + 'data' => ['example_key' => 'example_value'], + 'metadata' => ['tags' => ['finance'], '_uniqueId' => 'abc123'], 'config' => 'default', 'priority' => null, 'queue' => 'default', diff --git a/tests/TestCase/Listener/FailedJobsListenerTest.php b/tests/TestCase/Listener/FailedJobsListenerTest.php index a1c9bc2..0d13668 100644 --- a/tests/TestCase/Listener/FailedJobsListenerTest.php +++ b/tests/TestCase/Listener/FailedJobsListenerTest.php @@ -93,7 +93,7 @@ public function testFailedJobIsAddedWhenEventIsFired() $this->assertSame(LogToDebugJob::class, $failedJob->class); $this->assertSame('execute', $failedJob->method); - $this->assertSame(json_encode(['example_key' => 'example_value']), $failedJob->data); + $this->assertSame(['example_key' => 'example_value'], $failedJob->data); $this->assertSame('example_config', $failedJob->config); $this->assertSame('example_priority', $failedJob->priority); $this->assertSame('example_queue', $failedJob->queue); @@ -136,7 +136,7 @@ public function testFailedJobPreservesMetadata() $failedJob = $failedJobsTable->find()->first(); - $this->assertSame($parsedBody['metadata'], $failedJob->decoded_metadata); + $this->assertSame($parsedBody['metadata'], $failedJob->metadata); } public function testFailedJobWithoutMetadataStoresNull() @@ -175,7 +175,6 @@ public function testFailedJobWithoutMetadataStoresNull() $failedJob = $failedJobsTable->find()->first(); $this->assertNull($failedJob->metadata); - $this->assertSame([], $failedJob->decoded_metadata); } /** diff --git a/tests/schema.php b/tests/schema.php index 1dfba5f..9f65282 100644 --- a/tests/schema.php +++ b/tests/schema.php @@ -8,8 +8,8 @@ 'id' => ['type' => 'integer', 'length' => null, 'unsigned' => false, 'null' => false, 'default' => null, 'comment' => '', 'autoIncrement' => true, 'precision' => null], '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], - 'metadata' => ['type' => 'text', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null], + 'data' => ['type' => 'json', 'length' => null, 'null' => false, 'default' => null, 'comment' => '', 'precision' => null], + 'metadata' => ['type' => 'json', 'length' => null, '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], From 8b38f976ee938e4ca62038b68f8ce047a3652a86 Mon Sep 17 00:00:00 2001 From: Yevgeny Tomenko Date: Tue, 20 Oct 2026 19:02:12 +0300 Subject: [PATCH 3/3] Capture dtoClass and metadata in test transport --- src/TestSuite/QueueTrait.php | 60 +++++++++++++++++++++++++++++++ src/TestSuite/TestQueueClient.php | 11 ++++++ 2 files changed, 71 insertions(+) diff --git a/src/TestSuite/QueueTrait.php b/src/TestSuite/QueueTrait.php index 240771b..abef0dd 100644 --- a/src/TestSuite/QueueTrait.php +++ b/src/TestSuite/QueueTrait.php @@ -241,6 +241,66 @@ public function assertJobQueuedWithPriority( ); } + /** + * Assert a job was queued with specific metadata + * + * @param string $jobClass Job class name + * @param array $metadata Expected metadata + * @param string $message Optional assertion message + * @return void + */ + public function assertJobQueuedWithMetadata( + string $jobClass, + array $metadata, + string $message = '', + ): void { + $jobs = TestQueueClient::getQueuedJobsByClass($jobClass); + $this->assertNotEmpty($jobs, sprintf('Job %s was not queued', $jobClass)); + + $found = false; + foreach ($jobs as $job) { + if (($job['metadata'] ?? []) === $metadata) { + $found = true; + break; + } + } + + $this->assertTrue( + $found, + $message ?: sprintf('Job %s was not queued with expected metadata', $jobClass), + ); + } + + /** + * Assert a job was queued with a specific DTO class + * + * @param string $jobClass Job class name + * @param string $dtoClass Expected DTO class name + * @param string $message Optional assertion message + * @return void + */ + public function assertJobQueuedWithDtoClass( + string $jobClass, + string $dtoClass, + string $message = '', + ): void { + $jobs = TestQueueClient::getQueuedJobsByClass($jobClass); + $this->assertNotEmpty($jobs, sprintf('Job %s was not queued', $jobClass)); + + $found = false; + foreach ($jobs as $job) { + if (($job['dtoClass'] ?? null) === $dtoClass) { + $found = true; + break; + } + } + + $this->assertTrue( + $found, + $message ?: sprintf('Job %s was not queued with DTO class %s', $jobClass, $dtoClass), + ); + } + /** * Assert a job was queued a specific number of times * diff --git a/src/TestSuite/TestQueueClient.php b/src/TestSuite/TestQueueClient.php index 584a9dc..d02f7ee 100644 --- a/src/TestSuite/TestQueueClient.php +++ b/src/TestSuite/TestQueueClient.php @@ -108,6 +108,15 @@ public static function captureMessage( $jobClass = is_array($classData) ? $classData[0] : null; $method = is_array($classData) && isset($classData[1]) ? $classData[1] : 'execute'; $data = $body['data'] ?? ($body['args'][0] ?? []); + $dtoClass = $body['dtoClass'] ?? null; + if (!is_string($dtoClass)) { + $dtoClass = null; + } + + $metadata = $body['metadata'] ?? []; + if (!is_array($metadata)) { + $metadata = []; + } $requeueOptions = $body['requeueOptions'] ?? []; $configName = $requeueOptions['config'] ?? 'default'; @@ -145,6 +154,8 @@ public static function captureMessage( 'jobClass' => $jobClass, 'method' => $method, 'data' => $data, + 'dtoClass' => $dtoClass, + 'metadata' => $metadata, 'options' => [ 'config' => $configName, 'queue' => $queueName,