Fix queue tracing for non-object messages and when using AMQP (RabbitMQ)

[angelvilaplana]

Summary

This PR fixes an exception thrown when publishing messages to custom topics with non-array payloads.

Previously, the ExchangePlugin attempted to merge Sentry tracing data into the decoded message body using the PHP spread operator. When the payload was not an array (e.g. string or integer), PHP failed to unpack it and threw:

Error: Only arrays and Traversables can be unpacked

This PR changes how Sentry trace data is propagated by injecting it into the envelope properties instead of the message body. This avoids modifying or decoding the payload entirely and works with any payload type.

This PR solves the issue reported in #232

Result

• Publishing messages with non-array payloads no longer throws exceptions.
• Messages are enqueued and consumed normally with full tracing enabled (traces_sample_rate = 1.0)
• Sentry trace context is successfully propagated through envelope properties
• Message bodies are not decoded, modified, or re-encoded

Checklist

• I've ran composer run codestyle
• I've ran composer run analyse

[indykoning]

Thanks! What queueing engine do you use?
I remember wanting to add it to the properties before, but receiving errors as it didn't accept it.
But every queueing engine responds differently

[angelvilaplana]

Hi @indykoning, in the environments I've worked with, I've used RabbitMQ as the queue engine.
I also had problems when I tried to patch it to avoid this issue while running the queues. In my case, the problem occurred when I changed how I was setting the body.

After I fixed the "Error: Only arrays and Traversables can be unpacked" issue, a new problem appeared when running the convertMessage function inside the MessageEncoder class (https://github.com/magento/magento2/blob/2.4-develop/lib/internal/Magento/Framework/MessageQueue/MessageEncoder.php#L146-L162).

When trying to obtain the convertedMessage, it throws an exception depending on the data type of the queue. For example, if the data type is an object such as Magento\AsynchronousOperations\Api\Data\OperationInterface, an exception is thrown because SentryTrace is not supported, as it is not a property of that object.

Given the situation, I’ll change how the trace data is added to the envelope by setting it in the properties. This approach works correctly with AMQP.

However, after investigating the code further and testing the database queue engine, I discovered that once the queue reaches afterDequeue or beforeReject, the properties I previously added are no longer present.

Considering this, I refactored the code to avoid breaking the queue trace. I apologize for the mistake. I wasn’t aware that Magento supports more than one queue engine option.

Thank you very much for your comment, as it helped me test different scenarios and better understand why the variables are set in the body rather than in the properties.

[indykoning]

Running this using mysqlmq i'm running into the following error when running the consumers runner (magerun2 sys:cron:run consumers_runner)

Warning: Undefined array key "envelope_body" in vendor/justbetter/magento2-sentry/Plugin/Profiling/QueueMessageEncoderPlugin.php on line 48 (this can be seen in the system.log)

main.ERROR: Warning: Undefined array key "envelope_body" in vendor/justbetter/magento2-sentry/Plugin/Profiling/QueueMessageEncoderPlugin.php on line 48
#0 vendor/sentry/sentry/src/ErrorHandler.php(354): Magento\Framework\App\ErrorHandler->handler(2, 'Undefined array...', '...', 48, Array)
#1 vendor/justbetter/magento2-sentry/Plugin/Profiling/QueueMessageEncoderPlugin.php(48): Sentry\ErrorHandler->handleError(2, 'Undefined array...', '...', 48)
#2 vendor/magento/framework/Interception/Interceptor.php(121): JustBetter\Sentry\Plugin\Profiling\QueueMessageEncoderPlugin->beforeDecode(Object(Magento\Framework\MessageQueue\MessageEncoder\Interceptor), 'async.system.re...', '{"id":1,"bulk_u...')
#3 vendor/magento/framework/Interception/Interceptor.php(153): Magento\Framework\MessageQueue\MessageEncoder\Interceptor->Magento\Framework\Interception\{closure}('async.system.re...', '{"id":1,"bulk_u...')
#4 generated/code/Magento/Framework/MessageQueue/MessageEncoder/Interceptor.php(32): Magento\Framework\MessageQueue\MessageEncoder\Interceptor->___callPlugins('decode', Array, Array)
#5 vendor/magento/module-webapi-async/Plugin/AsynchronousOperations/MassConsumerEnvelopeCallback.php(90): Magento\Framework\MessageQueue\MessageEncoder\Interceptor->decode('async.system.re...', '{"id":1,"bulk_u...')
#6 vendor/magento/framework/Interception/Interceptor.php(135): Magento\WebapiAsync\Plugin\AsynchronousOperations\MassConsumerEnvelopeCallback->aroundExecute(Object(Magento\AsynchronousOperations\Model\MassConsumerEnvelopeCallback\Interceptor), Object(Closure), Object(Magento\Framework\MessageQueue\Envelope))
#7 vendor/magento/framework/Interception/Interceptor.php(153): Magento\AsynchronousOperations\Model\MassConsumerEnvelopeCallback\Interceptor->Magento\Framework\Interception\{closure}(Object(Magento\Framework\MessageQueue\Envelope))
#8 generated/code/Magento/AsynchronousOperations/Model/MassConsumerEnvelopeCallback/Interceptor.php(23): Magento\AsynchronousOperations\Model\MassConsumerEnvelopeCallback\Interceptor->___callPlugins('execute', Array, NULL)
#9 vendor/magento/module-asynchronous-operations/Model/MassConsumer.php(115): Magento\AsynchronousOperations\Model\MassConsumerEnvelopeCallback\Interceptor->execute(Object(Magento\Framework\MessageQueue\Envelope))
#10 vendor/magento/framework-message-queue/CallbackInvoker.php(100): Magento\AsynchronousOperations\Model\MassConsumer->Magento\AsynchronousOperations\Model\{closure}(Object(Magento\Framework\MessageQueue\Envelope))
#11 vendor/magento/module-asynchronous-operations/Model/MassConsumer.php(88): Magento\Framework\MessageQueue\CallbackInvoker->invoke(Object(Magento\MysqlMq\Model\Driver\Queue\Interceptor), '10000', Object(Closure), 9223372036854775807, 1)
#12 vendor/magento/module-message-queue/Console/StartConsumerCommand.php(94): Magento\AsynchronousOperations\Model\MassConsumer->process('10000')
#13 vendor/symfony/console/Command/Command.php(326): Magento\MessageQueue\Console\StartConsumerCommand->execute(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#14 vendor/magento/framework/Interception/Interceptor.php(58): Symfony\Component\Console\Command\Command->run(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#15 vendor/magento/framework/Interception/Interceptor.php(138): Magento\MessageQueue\Console\StartConsumerCommand\Interceptor->___callParent('run', Array)
#16 vendor/justbetter/magento2-sentry/Plugin/GlobalExceptionCatcher.php(112): Magento\MessageQueue\Console\StartConsumerCommand\Interceptor->Magento\Framework\Interception\{closure}(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#17 vendor/justbetter/magento2-sentry/Plugin/GlobalExceptionCatcher.php(78): JustBetter\Sentry\Plugin\GlobalExceptionCatcher->globalCatcher(Object(Magento\MessageQueue\Console\StartConsumerCommand\Interceptor), Object(Closure), Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#18 vendor/magento/framework/Interception/Interceptor.php(135): JustBetter\Sentry\Plugin\GlobalExceptionCatcher->aroundRun(Object(Magento\MessageQueue\Console\StartConsumerCommand\Interceptor), Object(Closure), Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#19 vendor/magento/framework/Interception/Interceptor.php(153): Magento\MessageQueue\Console\StartConsumerCommand\Interceptor->Magento\Framework\Interception\{closure}(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#20 generated/code/Magento/MessageQueue/Console/StartConsumerCommand/Interceptor.php(77): Magento\MessageQueue\Console\StartConsumerCommand\Interceptor->___callPlugins('run', Array, NULL)
#21 generated/code/Magento/MessageQueue/Console/StartConsumerCommand/Proxy.php(153): Magento\MessageQueue\Console\StartConsumerCommand\Interceptor->run(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#22 vendor/symfony/console/Application.php(1070): Magento\MessageQueue\Console\StartConsumerCommand\Proxy->run(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#23 vendor/symfony/console/Application.php(324): Symfony\Component\Console\Application->doRunCommand(Object(Magento\MessageQueue\Console\StartConsumerCommand\Proxy), Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#24 vendor/magento/framework/Console/Cli.php(122): Symfony\Component\Console\Application->doRun(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#25 vendor/symfony/console/Application.php(175): Magento\Framework\Console\Cli->doRun(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#26 bin/magento(23): Symfony\Component\Console\Application->run()
#27 {main} [] []

[angelvilaplana]

Hi @indykoning, thank you for your report. To be honest, I had a lot of trouble trying to reproduce the issue because I couldn't reproduce it at first. In the end, I noticed that if I publish a queue with the previous changes and then run the consumer with the new changes, I get the same problem you experienced. That allowed me to reproduce it and fix it.