From 6165c70e1b271588d71b6db27a5a7ca9a5fe6441 Mon Sep 17 00:00:00 2001 From: Carl Schwan Date: Sun, 4 Oct 2026 22:34:29 +0200 Subject: [PATCH] feat(core): add a message queue for asynchronous tasks Apps declare messages with #[AsMessage] and handlers with section of info.xml and dispatch messages through IMessageBus. Messages are stored in the database and consumed after the background jobs of each cron run, so no additional software is needed. Large instances can disable this with 'message_queue.consume_in_cron' and run dedicated workers with occ message-queue:consume. When running under PHP-FPM, messages dispatched during a web request are handled by the same process after the response was sent with fastcgi_finish_request(), so small instances don't have to wait for the next cron run. Assisted-by: ClaudeCode:claude-opus-5-5 Signed-off-by: Carl Schwan --- build/rector-strict.php | 13 + config/config.sample.php | 26 ++ core/Command/MessageQueue/Consume.php | 84 +++++ .../Version36000Date20261004120000.php | 62 ++++ core/Service/CronService.php | 29 ++ core/register_command.php | 1 + lib/composer/composer/autoload_classmap.php | 17 + lib/composer/composer/autoload_static.php | 17 + lib/private/App/InfoParser.php | 6 + .../MessageQueue/AfterRequestHandler.php | 96 +++++ lib/private/MessageQueue/Consumer.php | 269 ++++++++++++++ lib/private/MessageQueue/HandlerRegistry.php | 125 +++++++ lib/private/MessageQueue/MessageBus.php | 45 +++ .../MessageQueue/MessageMetadataReader.php | 48 +++ .../MessageQueue/MessageNormalizer.php | 142 ++++++++ lib/private/MessageQueue/MessageStore.php | 223 ++++++++++++ lib/private/MessageQueue/QueuedMessage.php | 23 ++ lib/private/Server.php | 3 + lib/public/App/AppInfoDefinition.php | 1 + .../MessageQueue/Attribute/AsMessage.php | 60 +++ .../Attribute/AsMessageHandler.php | 55 +++ .../Exception/InvalidMessageException.php | 19 + .../Exception/RecoverableMessageException.php | 20 + .../UnrecoverableMessageException.php | 19 + lib/public/MessageQueue/IMessageBus.php | 33 ++ lib/public/MessageQueue/Queue.php | 35 ++ psalm-strict.xml | 4 + resources/app-info-shipped.xsd | 9 + resources/app-info.xsd | 9 + tests/data/app/appinfo-attributes-once.json | 1 + tests/data/app/appinfo-multi-once.json | 1 + tests/data/app/appinfo-multi-twice.json | 1 + tests/data/app/expected-info.json | 1 + tests/data/app/navigation-one-item.json | 1 + tests/data/app/navigation-two-items.json | 1 + tests/data/app/various-single-item.json | 1 + .../MessageQueue/AfterRequestHandlerTest.php | 92 +++++ .../Fixtures/DuplicateHandler.php | 20 + .../MessageQueue/Fixtures/FailingMessage.php | 19 + .../Fixtures/InvokableHandler.php | 20 + .../lib/MessageQueue/Fixtures/LowMessage.php | 19 + .../MessageQueue/Fixtures/MultiHandler.php | 52 +++ .../MessageQueue/Fixtures/NestedMessage.php | 18 + .../lib/MessageQueue/Fixtures/NotAMessage.php | 15 + tests/lib/MessageQueue/Fixtures/Status.php | 13 + .../Fixtures/TestAfterRequestHandler.php | 35 ++ .../lib/MessageQueue/Fixtures/TestMessage.php | 25 ++ .../lib/MessageQueue/Fixtures/UserMessage.php | 18 + .../lib/MessageQueue/HandlerRegistryTest.php | 79 ++++ .../MessageQueue/MessageNormalizerTest.php | 84 +++++ tests/lib/MessageQueue/MessageQueueTest.php | 342 ++++++++++++++++++ version.php | 2 +- 52 files changed, 2352 insertions(+), 1 deletion(-) create mode 100644 core/Command/MessageQueue/Consume.php create mode 100644 core/Migrations/Version36000Date20261004120000.php create mode 100644 lib/private/MessageQueue/AfterRequestHandler.php create mode 100644 lib/private/MessageQueue/Consumer.php create mode 100644 lib/private/MessageQueue/HandlerRegistry.php create mode 100644 lib/private/MessageQueue/MessageBus.php create mode 100644 lib/private/MessageQueue/MessageMetadataReader.php create mode 100644 lib/private/MessageQueue/MessageNormalizer.php create mode 100644 lib/private/MessageQueue/MessageStore.php create mode 100644 lib/private/MessageQueue/QueuedMessage.php create mode 100644 lib/public/MessageQueue/Attribute/AsMessage.php create mode 100644 lib/public/MessageQueue/Attribute/AsMessageHandler.php create mode 100644 lib/public/MessageQueue/Exception/InvalidMessageException.php create mode 100644 lib/public/MessageQueue/Exception/RecoverableMessageException.php create mode 100644 lib/public/MessageQueue/Exception/UnrecoverableMessageException.php create mode 100644 lib/public/MessageQueue/IMessageBus.php create mode 100644 lib/public/MessageQueue/Queue.php create mode 100644 tests/lib/MessageQueue/AfterRequestHandlerTest.php create mode 100644 tests/lib/MessageQueue/Fixtures/DuplicateHandler.php create mode 100644 tests/lib/MessageQueue/Fixtures/FailingMessage.php create mode 100644 tests/lib/MessageQueue/Fixtures/InvokableHandler.php create mode 100644 tests/lib/MessageQueue/Fixtures/LowMessage.php create mode 100644 tests/lib/MessageQueue/Fixtures/MultiHandler.php create mode 100644 tests/lib/MessageQueue/Fixtures/NestedMessage.php create mode 100644 tests/lib/MessageQueue/Fixtures/NotAMessage.php create mode 100644 tests/lib/MessageQueue/Fixtures/Status.php create mode 100644 tests/lib/MessageQueue/Fixtures/TestAfterRequestHandler.php create mode 100644 tests/lib/MessageQueue/Fixtures/TestMessage.php create mode 100644 tests/lib/MessageQueue/Fixtures/UserMessage.php create mode 100644 tests/lib/MessageQueue/HandlerRegistryTest.php create mode 100644 tests/lib/MessageQueue/MessageNormalizerTest.php create mode 100644 tests/lib/MessageQueue/MessageQueueTest.php diff --git a/build/rector-strict.php b/build/rector-strict.php index a4457f40dc22d..a28dab65968a5 100644 --- a/build/rector-strict.php +++ b/build/rector-strict.php @@ -6,7 +6,9 @@ */ use Rector\DeadCode\Rector\ClassMethod\RemoveDuplicatedReturnSelfDocblockRector; +use Rector\DeadCode\Rector\ClassMethod\RemoveEmptyClassMethodRector; use Rector\DeadCode\Rector\ClassMethod\RemoveReturnTagIncompatibleWithNativeTypeRector; +use Rector\DeadCode\Rector\ClassMethod\RemoveUnusedPublicMethodParameterRector; use Rector\Php81\Rector\Property\ReadOnlyPropertyRector; use Rector\Php82\Rector\Class_\ReadOnlyClassRector; use Rector\PHPUnit\CodeQuality\Rector\Class_\AddSeeTestAnnotationRector; @@ -63,6 +65,10 @@ $nextcloudDir . '/lib/public/SystemReport', $nextcloudDir . '/lib/private/SystemReport', $nextcloudDir . '/tests/lib/SystemReport', + $nextcloudDir . '/lib/public/MessageQueue', + $nextcloudDir . '/lib/private/MessageQueue', + $nextcloudDir . '/core/Command/MessageQueue', + $nextcloudDir . '/tests/lib/MessageQueue', ]) ->withAutoloadPaths([ // ensure rector properly autoload the public interfaces @@ -102,4 +108,11 @@ // non-final type; removing it breaks psalm's // LessSpecificImplementedReturnType check (psalm-strict). RemoveDuplicatedReturnSelfDocblockRector::class, + // Message handler fixtures declare the handled message through their parameter type + RemoveUnusedPublicMethodParameterRector::class => [ + $nextcloudDir . '/tests/lib/MessageQueue/Fixtures', + ], + RemoveEmptyClassMethodRector::class => [ + $nextcloudDir . '/tests/lib/MessageQueue/Fixtures', + ], ]); diff --git a/config/config.sample.php b/config/config.sample.php index c75c85a433718..e9988e3abda4e 100644 --- a/config/config.sample.php +++ b/config/config.sample.php @@ -1767,6 +1767,7 @@ * the background jobs which advertise themselves as not time sensitive will be * delayed during the "working" hours and only run in the 4 hours after the given time. * This is, e.g., used for activity expiration, suspicious login training, and update checks. + * Messages in the low priority queue of the message queue are delayed the same way. * * A value of 1, e.g., will only run these background jobs between 01:00am UTC and 05:00am UTC. * @@ -1774,6 +1775,31 @@ */ 'maintenance_window_start' => 1, + /** + * Consume the message queue during cron runs + * + * Messages dispatched by apps are handled after the background jobs of each + * cron run. Large instances can disable this and instead run one or more + * dedicated workers with ``occ message-queue:consume``. + * + * Defaults to ``true`` + */ + 'message_queue.consume_in_cron' => true, + + /** + * Handle messages right after the request + * + * When running under PHP-FPM, messages dispatched during a web request are + * handled by the same PHP process after the response was sent to the + * client, so they don't have to wait for the next cron run. Messages that + * are not handled within a few seconds stay in the queue. Large instances + * running dedicated workers can disable this to keep PHP-FPM processes + * available for requests. + * + * Defaults to ``true`` + */ + 'message_queue.handle_after_request' => true, + /** * Log all LDAP requests into a file * diff --git a/core/Command/MessageQueue/Consume.php b/core/Command/MessageQueue/Consume.php new file mode 100644 index 0000000000000..1f6ccc72e56e2 --- /dev/null +++ b/core/Command/MessageQueue/Consume.php @@ -0,0 +1,84 @@ +setName('message-queue:consume') + ->setDescription('Run a worker consuming messages from the message queue') + ->addArgument( + 'queues', + InputArgument::OPTIONAL | InputArgument::IS_ARRAY, + 'Queues to consume, in order of priority (' . implode(', ', array_map(static fn (Queue $queue) => $queue->value, Queue::cases())) . ')', + ) + ->addOption('time-limit', 't', InputOption::VALUE_REQUIRED, 'Stop after this many seconds') + ->addOption('limit', 'l', InputOption::VALUE_REQUIRED, 'Stop after handling this many messages') + ->addOption('memory-limit', 'm', InputOption::VALUE_REQUIRED, 'Stop when the memory usage exceeds this limit, e.g. 512M') + ->addOption('sleep', null, InputOption::VALUE_REQUIRED, 'Seconds to wait before polling again when all queues are empty', '1') + ->addOption('stop-when-empty', null, InputOption::VALUE_NONE, 'Stop as soon as all queues are empty'); + } + + #[\Override] + protected function execute(InputInterface $input, OutputInterface $output): int { + /** @var list $names */ + $names = $input->getArgument('queues'); + $queues = []; + foreach ($names as $name) { + $queue = Queue::tryFrom($name); + if ($queue === null) { + $output->writeln('Unknown queue ' . $name . ''); + return self::FAILURE; + } + + $queues[] = $queue; + } + + /** @var mixed $memoryLimit */ + $memoryLimit = $input->getOption('memory-limit'); + $this->consumer->consume( + queues: $queues ?: Queue::cases(), + timeLimit: $this->getIntOption($input, 'time-limit'), + messageLimit: $this->getIntOption($input, 'limit'), + memoryLimit: is_string($memoryLimit) ? (int)Util::computerFileSize($memoryLimit) : null, + stopWhenEmpty: (bool)$input->getOption('stop-when-empty'), + sleep: (int)$input->getOption('sleep'), + output: $output->isVerbose() ? static fn (string $message) => $output->writeln($message) : null, + ); + + return self::SUCCESS; + } + + #[\Override] + public function cancelOperation(): void { + parent::cancelOperation(); + $this->consumer->stop(); + } + + private function getIntOption(InputInterface $input, string $name): ?int { + /** @var mixed $value */ + $value = $input->getOption($name); + return is_string($value) ? (int)$value : null; + } +} diff --git a/core/Migrations/Version36000Date20261004120000.php b/core/Migrations/Version36000Date20261004120000.php new file mode 100644 index 0000000000000..6404bc72375cf --- /dev/null +++ b/core/Migrations/Version36000Date20261004120000.php @@ -0,0 +1,62 @@ +hasTable('message_queue')) { + $table = $schema->createTable('message_queue'); + $table->addColumn('id', Types::BIGINT, [ + 'notnull' => true, + 'unsigned' => true, + ]); + $table->addColumn('queue_name', Types::STRING, ['notnull' => true, 'length' => 32]); + $table->addColumn('message_class', Types::STRING, ['notnull' => true, 'length' => 255]); + $table->addColumn('body', Types::TEXT, ['notnull' => true]); + $table->addColumn('retry_count', Types::INTEGER, ['notnull' => true, 'default' => 0, 'unsigned' => true]); + $table->addColumn('last_error', Types::TEXT, ['notnull' => false]); + $table->addColumn('created_at', Types::BIGINT, ['notnull' => true, 'unsigned' => true]); + $table->addColumn('available_at', Types::BIGINT, ['notnull' => true, 'unsigned' => true]); + $table->addColumn('delivered_at', Types::BIGINT, ['notnull' => false, 'unsigned' => true]); + $table->addColumn('deduplication_hash', Types::STRING, ['notnull' => false, 'length' => 64]); + $table->setPrimaryKey(['id']); + $table->addIndex(['queue_name', 'available_at'], 'mq_queue_available'); + $table->addIndex(['deduplication_hash'], 'mq_dedup_hash'); + // Makes sure there is no auto-increment in Oracle + $schema->dropAutoincrementColumn('message_queue', 'id'); + + return $schema; + } + + return null; + } +} diff --git a/core/Service/CronService.php b/core/Service/CronService.php index 379b60b10298f..66fb734c54194 100644 --- a/core/Service/CronService.php +++ b/core/Service/CronService.php @@ -16,6 +16,7 @@ use OC\BackgroundJob\JobClassesRegistry; use OC\BackgroundJob\JobRuns; use OC\DB\Connection; +use OC\MessageQueue\Consumer; use OC\Security\CSRF\TokenStorage\SessionStorage; use OC\Session\CryptoWrapper; use OC\Session\Memory; @@ -29,6 +30,7 @@ use OCP\ILogger; use OCP\ISession; use OCP\ITempManager; +use OCP\MessageQueue\Queue; use OCP\Util; use Psr\Log\LoggerInterface; @@ -54,6 +56,7 @@ public function __construct( private readonly JobRuns $jobRuns, private readonly JobClassesRegistry $jobClassesRegistry, private readonly ISetupManager $setupManager, + private readonly Consumer $messageQueueConsumer, private readonly bool $isCLI, ) { } @@ -271,6 +274,11 @@ private function runCli(string $appMode, ?array $jobClasses): void { } } + if ($jobClasses === null) { + $queues = $onlyTimeSensitive ? [Queue::High, Queue::Default] : Queue::cases(); + $this->consumeMessages($queues, max(60, $endTime - time())); + } + // Makes sure last error isn't caught by shutdown function error_clear_last(); } @@ -287,6 +295,27 @@ private function runWeb(string $appMode): void { $job->start($this->jobList); $this->jobList->setLastJob($job); } + $this->consumeMessages(Queue::cases(), 10); + } + } + + /** + * @param list $queues + */ + private function consumeMessages(array $queues, int $timeLimit): void { + if (!$this->config->getSystemValueBool('message_queue.consume_in_cron', true)) { + return; + } + + try { + $this->messageQueueConsumer->consume( + queues: $queues, + timeLimit: $timeLimit, + stopWhenEmpty: true, + output: $this->verboseCallback, + ); + } catch (\Throwable $e) { + $this->logger->error('Error while consuming the message queue: ' . $e->getMessage(), ['app' => 'cron', 'exception' => $e]); } } diff --git a/core/register_command.php b/core/register_command.php index d5f8d76279330..27c4acc786f07 100644 --- a/core/register_command.php +++ b/core/register_command.php @@ -156,6 +156,7 @@ $application->addCommand(Server::get(Delete::class)); $application->addCommand(Server::get(JobWorker::class)); $application->addCommand(Server::get(RunningJobs::class)); + $application->addCommand(Server::get(Command\MessageQueue\Consume::class)); $application->addCommand(Server::get(JobsHistory::class)); $application->addCommand(Server::get(Test::class)); diff --git a/lib/composer/composer/autoload_classmap.php b/lib/composer/composer/autoload_classmap.php index a641ddddb350b..58302ae1e84a8 100644 --- a/lib/composer/composer/autoload_classmap.php +++ b/lib/composer/composer/autoload_classmap.php @@ -784,6 +784,13 @@ 'OCP\\Mail\\Provider\\IProvider' => $baseDir . '/lib/public/Mail/Provider/IProvider.php', 'OCP\\Mail\\Provider\\IService' => $baseDir . '/lib/public/Mail/Provider/IService.php', 'OCP\\Mail\\Provider\\Message' => $baseDir . '/lib/public/Mail/Provider/Message.php', + 'OCP\\MessageQueue\\Attribute\\AsMessage' => $baseDir . '/lib/public/MessageQueue/Attribute/AsMessage.php', + 'OCP\\MessageQueue\\Attribute\\AsMessageHandler' => $baseDir . '/lib/public/MessageQueue/Attribute/AsMessageHandler.php', + 'OCP\\MessageQueue\\Exception\\InvalidMessageException' => $baseDir . '/lib/public/MessageQueue/Exception/InvalidMessageException.php', + 'OCP\\MessageQueue\\Exception\\RecoverableMessageException' => $baseDir . '/lib/public/MessageQueue/Exception/RecoverableMessageException.php', + 'OCP\\MessageQueue\\Exception\\UnrecoverableMessageException' => $baseDir . '/lib/public/MessageQueue/Exception/UnrecoverableMessageException.php', + 'OCP\\MessageQueue\\IMessageBus' => $baseDir . '/lib/public/MessageQueue/IMessageBus.php', + 'OCP\\MessageQueue\\Queue' => $baseDir . '/lib/public/MessageQueue/Queue.php', 'OCP\\Migration\\Attributes\\AddColumn' => $baseDir . '/lib/public/Migration/Attributes/AddColumn.php', 'OCP\\Migration\\Attributes\\AddIndex' => $baseDir . '/lib/public/Migration/Attributes/AddIndex.php', 'OCP\\Migration\\Attributes\\ColumnMigrationAttribute' => $baseDir . '/lib/public/Migration/Attributes/ColumnMigrationAttribute.php', @@ -1534,6 +1541,7 @@ 'OC\\Core\\Command\\Memcache\\DistributedGet' => $baseDir . '/core/Command/Memcache/DistributedGet.php', 'OC\\Core\\Command\\Memcache\\DistributedSet' => $baseDir . '/core/Command/Memcache/DistributedSet.php', 'OC\\Core\\Command\\Memcache\\RedisCommand' => $baseDir . '/core/Command/Memcache/RedisCommand.php', + 'OC\\Core\\Command\\MessageQueue\\Consume' => $baseDir . '/core/Command/MessageQueue/Consume.php', 'OC\\Core\\Command\\OCM\\ActivateKey' => $baseDir . '/core/Command/OCM/ActivateKey.php', 'OC\\Core\\Command\\OCM\\ListKeys' => $baseDir . '/core/Command/OCM/ListKeys.php', 'OC\\Core\\Command\\OCM\\RetireKey' => $baseDir . '/core/Command/OCM/RetireKey.php', @@ -1743,6 +1751,7 @@ 'OC\\Core\\Migrations\\Version34000Date20260518163022' => $baseDir . '/core/Migrations/Version34000Date20260518163022.php', 'OC\\Core\\Migrations\\Version34000Date20260521110333' => $baseDir . '/core/Migrations/Version34000Date20260521110333.php', 'OC\\Core\\Migrations\\Version35000Date20260527162338' => $baseDir . '/core/Migrations/Version35000Date20260527162338.php', + 'OC\\Core\\Migrations\\Version36000Date20261004120000' => $baseDir . '/core/Migrations/Version36000Date20261004120000.php', 'OC\\Core\\Notification\\CoreNotifier' => $baseDir . '/core/Notification/CoreNotifier.php', 'OC\\Core\\ResponseDefinitions' => $baseDir . '/core/ResponseDefinitions.php', 'OC\\Core\\Service\\CronService' => $baseDir . '/core/Service/CronService.php', @@ -2098,6 +2107,14 @@ 'OC\\Memcache\\Redis' => $baseDir . '/lib/private/Memcache/Redis.php', 'OC\\Memcache\\WithLocalCache' => $baseDir . '/lib/private/Memcache/WithLocalCache.php', 'OC\\MemoryInfo' => $baseDir . '/lib/private/MemoryInfo.php', + 'OC\\MessageQueue\\AfterRequestHandler' => $baseDir . '/lib/private/MessageQueue/AfterRequestHandler.php', + 'OC\\MessageQueue\\Consumer' => $baseDir . '/lib/private/MessageQueue/Consumer.php', + 'OC\\MessageQueue\\HandlerRegistry' => $baseDir . '/lib/private/MessageQueue/HandlerRegistry.php', + 'OC\\MessageQueue\\MessageBus' => $baseDir . '/lib/private/MessageQueue/MessageBus.php', + 'OC\\MessageQueue\\MessageMetadataReader' => $baseDir . '/lib/private/MessageQueue/MessageMetadataReader.php', + 'OC\\MessageQueue\\MessageNormalizer' => $baseDir . '/lib/private/MessageQueue/MessageNormalizer.php', + 'OC\\MessageQueue\\MessageStore' => $baseDir . '/lib/private/MessageQueue/MessageStore.php', + 'OC\\MessageQueue\\QueuedMessage' => $baseDir . '/lib/private/MessageQueue/QueuedMessage.php', 'OC\\Migration\\BackgroundRepair' => $baseDir . '/lib/private/Migration/BackgroundRepair.php', 'OC\\Migration\\ConsoleOutput' => $baseDir . '/lib/private/Migration/ConsoleOutput.php', 'OC\\Migration\\Exceptions\\AttributeException' => $baseDir . '/lib/private/Migration/Exceptions/AttributeException.php', diff --git a/lib/composer/composer/autoload_static.php b/lib/composer/composer/autoload_static.php index 1931ad3bff8a3..f0824192967bc 100644 --- a/lib/composer/composer/autoload_static.php +++ b/lib/composer/composer/autoload_static.php @@ -830,6 +830,13 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2 'OCP\\Mail\\Provider\\IProvider' => __DIR__ . '/../../..' . '/lib/public/Mail/Provider/IProvider.php', 'OCP\\Mail\\Provider\\IService' => __DIR__ . '/../../..' . '/lib/public/Mail/Provider/IService.php', 'OCP\\Mail\\Provider\\Message' => __DIR__ . '/../../..' . '/lib/public/Mail/Provider/Message.php', + 'OCP\\MessageQueue\\Attribute\\AsMessage' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Attribute/AsMessage.php', + 'OCP\\MessageQueue\\Attribute\\AsMessageHandler' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Attribute/AsMessageHandler.php', + 'OCP\\MessageQueue\\Exception\\InvalidMessageException' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Exception/InvalidMessageException.php', + 'OCP\\MessageQueue\\Exception\\RecoverableMessageException' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Exception/RecoverableMessageException.php', + 'OCP\\MessageQueue\\Exception\\UnrecoverableMessageException' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Exception/UnrecoverableMessageException.php', + 'OCP\\MessageQueue\\IMessageBus' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/IMessageBus.php', + 'OCP\\MessageQueue\\Queue' => __DIR__ . '/../../..' . '/lib/public/MessageQueue/Queue.php', 'OCP\\Migration\\Attributes\\AddColumn' => __DIR__ . '/../../..' . '/lib/public/Migration/Attributes/AddColumn.php', 'OCP\\Migration\\Attributes\\AddIndex' => __DIR__ . '/../../..' . '/lib/public/Migration/Attributes/AddIndex.php', 'OCP\\Migration\\Attributes\\ColumnMigrationAttribute' => __DIR__ . '/../../..' . '/lib/public/Migration/Attributes/ColumnMigrationAttribute.php', @@ -1580,6 +1587,7 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2 'OC\\Core\\Command\\Memcache\\DistributedGet' => __DIR__ . '/../../..' . '/core/Command/Memcache/DistributedGet.php', 'OC\\Core\\Command\\Memcache\\DistributedSet' => __DIR__ . '/../../..' . '/core/Command/Memcache/DistributedSet.php', 'OC\\Core\\Command\\Memcache\\RedisCommand' => __DIR__ . '/../../..' . '/core/Command/Memcache/RedisCommand.php', + 'OC\\Core\\Command\\MessageQueue\\Consume' => __DIR__ . '/../../..' . '/core/Command/MessageQueue/Consume.php', 'OC\\Core\\Command\\OCM\\ActivateKey' => __DIR__ . '/../../..' . '/core/Command/OCM/ActivateKey.php', 'OC\\Core\\Command\\OCM\\ListKeys' => __DIR__ . '/../../..' . '/core/Command/OCM/ListKeys.php', 'OC\\Core\\Command\\OCM\\RetireKey' => __DIR__ . '/../../..' . '/core/Command/OCM/RetireKey.php', @@ -1789,6 +1797,7 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2 'OC\\Core\\Migrations\\Version34000Date20260518163022' => __DIR__ . '/../../..' . '/core/Migrations/Version34000Date20260518163022.php', 'OC\\Core\\Migrations\\Version34000Date20260521110333' => __DIR__ . '/../../..' . '/core/Migrations/Version34000Date20260521110333.php', 'OC\\Core\\Migrations\\Version35000Date20260527162338' => __DIR__ . '/../../..' . '/core/Migrations/Version35000Date20260527162338.php', + 'OC\\Core\\Migrations\\Version36000Date20261004120000' => __DIR__ . '/../../..' . '/core/Migrations/Version36000Date20261004120000.php', 'OC\\Core\\Notification\\CoreNotifier' => __DIR__ . '/../../..' . '/core/Notification/CoreNotifier.php', 'OC\\Core\\ResponseDefinitions' => __DIR__ . '/../../..' . '/core/ResponseDefinitions.php', 'OC\\Core\\Service\\CronService' => __DIR__ . '/../../..' . '/core/Service/CronService.php', @@ -2144,6 +2153,14 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2 'OC\\Memcache\\Redis' => __DIR__ . '/../../..' . '/lib/private/Memcache/Redis.php', 'OC\\Memcache\\WithLocalCache' => __DIR__ . '/../../..' . '/lib/private/Memcache/WithLocalCache.php', 'OC\\MemoryInfo' => __DIR__ . '/../../..' . '/lib/private/MemoryInfo.php', + 'OC\\MessageQueue\\AfterRequestHandler' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/AfterRequestHandler.php', + 'OC\\MessageQueue\\Consumer' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/Consumer.php', + 'OC\\MessageQueue\\HandlerRegistry' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/HandlerRegistry.php', + 'OC\\MessageQueue\\MessageBus' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/MessageBus.php', + 'OC\\MessageQueue\\MessageMetadataReader' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/MessageMetadataReader.php', + 'OC\\MessageQueue\\MessageNormalizer' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/MessageNormalizer.php', + 'OC\\MessageQueue\\MessageStore' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/MessageStore.php', + 'OC\\MessageQueue\\QueuedMessage' => __DIR__ . '/../../..' . '/lib/private/MessageQueue/QueuedMessage.php', 'OC\\Migration\\BackgroundRepair' => __DIR__ . '/../../..' . '/lib/private/Migration/BackgroundRepair.php', 'OC\\Migration\\ConsoleOutput' => __DIR__ . '/../../..' . '/lib/private/Migration/ConsoleOutput.php', 'OC\\Migration\\Exceptions\\AttributeException' => __DIR__ . '/../../..' . '/lib/private/Migration/Exceptions/AttributeException.php', diff --git a/lib/private/App/InfoParser.php b/lib/private/App/InfoParser.php index 94e8951fd372e..66b12c2685482 100644 --- a/lib/private/App/InfoParser.php +++ b/lib/private/App/InfoParser.php @@ -93,6 +93,9 @@ public function parse(string $file): ?array { if (!array_key_exists('commands', $array)) { $array['commands'] = []; } + if (!array_key_exists('message-handlers', $array)) { + $array['message-handlers'] = []; + } if (!array_key_exists('activity', $array)) { $array['activity'] = []; } @@ -160,6 +163,9 @@ public function parse(string $file): ?array { if (isset($array['commands']['command']) && is_array($array['commands']['command'])) { $array['commands'] = $array['commands']['command']; } + if (isset($array['message-handlers']['handler']) && is_array($array['message-handlers']['handler'])) { + $array['message-handlers'] = $array['message-handlers']['handler']; + } if (isset($array['two-factor-providers']['provider']) && is_array($array['two-factor-providers']['provider'])) { $array['two-factor-providers'] = $array['two-factor-providers']['provider']; } diff --git a/lib/private/MessageQueue/AfterRequestHandler.php b/lib/private/MessageQueue/AfterRequestHandler.php new file mode 100644 index 0000000000000..eb3e4204c8441 --- /dev/null +++ b/lib/private/MessageQueue/AfterRequestHandler.php @@ -0,0 +1,96 @@ + */ + private array $messageIds = []; + + private bool $registered = false; + + public function __construct( + private readonly Consumer $consumer, + private readonly IConfig $config, + private readonly ISession $session, + private readonly IUserSession $userSession, + private readonly ISetupManager $setupManager, + private readonly ILockingProvider $lockingProvider, + private readonly LoggerInterface $logger, + private readonly bool $isCLI, + ) { + } + + public function add(Queue $queue, string $messageId): void { + if ($queue === Queue::Low || !$this->isEnabled()) { + return; + } + + $this->messageIds[] = $messageId; + if (!$this->registered) { + $this->registered = true; + $this->registerShutdownFunction($this->run(...)); + } + } + + public function run(): void { + $messageIds = $this->messageIds; + $this->messageIds = []; + if ($messageIds === []) { + return; + } + + try { + $this->session->close(); + $this->finishRequest(); + + $this->userSession->setVolatileActiveUser(null); + $this->setupManager->tearDown(); + $this->consumer->consumeMessages($messageIds, self::TIME_LIMIT, self::MESSAGE_LIMIT); + } catch (\Throwable $throwable) { + $this->logger->error('Error while handling messages after the request: ' . $throwable->getMessage(), ['exception' => $throwable]); + } finally { + // The shutdown functions registered during boot already ran + $this->lockingProvider->releaseAll(); + } + } + + protected function isEnabled(): bool { + return !$this->isCLI + && $this->canFinishRequest() + && $this->config->getSystemValueBool('message_queue.handle_after_request', true); + } + + protected function canFinishRequest(): bool { + return function_exists('fastcgi_finish_request'); + } + + protected function registerShutdownFunction(callable $callback): void { + register_shutdown_function($callback); + } + + protected function finishRequest(): void { + fastcgi_finish_request(); + } +} diff --git a/lib/private/MessageQueue/Consumer.php b/lib/private/MessageQueue/Consumer.php new file mode 100644 index 0000000000000..01fba64445fd6 --- /dev/null +++ b/lib/private/MessageQueue/Consumer.php @@ -0,0 +1,269 @@ + $queues + * @param int|null $timeLimit Stop after this many seconds + * @param int|null $messageLimit Stop after handling this many messages + * @param int|null $memoryLimit Stop when the memory usage exceeds this many bytes + * @param bool $stopWhenEmpty Stop as soon as all queues are empty instead of waiting for new messages + * @param int $sleep Seconds to wait before polling again when all queues are empty + * @param (callable(string): void)|null $output Receives progress messages + */ + public function consume( + array $queues = [Queue::High, Queue::Default, Queue::Low], + ?int $timeLimit = null, + ?int $messageLimit = null, + ?int $memoryLimit = null, + bool $stopWhenEmpty = false, + int $sleep = 1, + ?callable $output = null, + ): void { + $next = function () use ($queues): ?QueuedMessage { + foreach ($queues as $queue) { + $message = $this->store->claimNext($queue); + if ($message instanceof QueuedMessage) { + return $message; + } + } + + return null; + }; + $this->run($next, $timeLimit, $messageLimit, $memoryLimit, $stopWhenEmpty, $sleep, $output); + } + + /** + * Handles the given messages in order, if they are still available. + * + * @param list $messageIds + */ + public function consumeMessages(array $messageIds, ?int $timeLimit = null, ?int $messageLimit = null): void { + $next = function () use (&$messageIds): ?QueuedMessage { + while (($id = array_shift($messageIds)) !== null) { + $message = $this->store->claimById($id); + if ($message instanceof QueuedMessage) { + return $message; + } + } + + return null; + }; + $this->run($next, $timeLimit, $messageLimit, stopWhenEmpty: true); + } + + /** + * Stops consuming after the current message. + */ + public function stop(): void { + $this->shouldStop = true; + } + + /** + * @param \Closure(): ?QueuedMessage $next + * @param (callable(string): void)|null $output + */ + private function run( + \Closure $next, + ?int $timeLimit = null, + ?int $messageLimit = null, + ?int $memoryLimit = null, + bool $stopWhenEmpty = false, + int $sleep = 1, + ?callable $output = null, + ): void { + $endTime = $timeLimit !== null ? microtime(true) + (float)$timeLimit : null; + $handled = 0; + + try { + while (!$this->shouldStop) { + if ($endTime !== null && microtime(true) >= $endTime) { + break; + } + + $message = $next(); + if ($message instanceof QueuedMessage) { + $this->handle($message, $output); + ++$handled; + if ($messageLimit !== null && $handled >= $messageLimit) { + break; + } + + if ($memoryLimit !== null && memory_get_usage(true) > $memoryLimit) { + break; + } + } elseif ($stopWhenEmpty) { + break; + } else { + sleep($sleep); + } + + $this->dispatchSignals(); + } + } finally { + $this->shouldStop = false; + } + } + + /** + * Runs the signal handlers, which may call stop(). + */ + private function dispatchSignals(): void { + if (function_exists('pcntl_signal_dispatch')) { + pcntl_signal_dispatch(); + } + } + + /** + * @param (callable(string): void)|null $output + */ + private function handle(QueuedMessage $queued, ?callable $output): void { + if ($output !== null) { + $output('Handling ' . $queued->messageClass . ' (' . $queued->id . ')'); + } + + try { + $this->invokeHandler($queued); + $error = null; + } catch (\Throwable $throwable) { + $error = $throwable; + } + + // Before writing to the store, so a transaction left open by the handler can't roll it back + $this->cleanUp(); + + if (!$error instanceof \Throwable) { + $this->store->delete($queued->id); + return; + } + + $delay = $this->getRetryDelay($queued, $error); + if ($delay !== null) { + $this->logger->warning('Error while handling message ' . $queued->messageClass . ', retrying in ' . $delay . ' seconds: ' . $error->getMessage(), ['exception' => $error]); + $this->store->retry($queued->id, $delay); + } else { + $this->logger->error('Error while handling message ' . $queued->messageClass . ', moving it to the failed queue: ' . $error->getMessage(), ['exception' => $error]); + $this->store->fail($queued->id, $error::class . ': ' . $error->getMessage()); + } + + if ($output !== null) { + $output('Failed ' . $queued->messageClass . ': ' . $error->getMessage() . ($delay !== null ? ' (will retry)' : '')); + } + } + + private function invokeHandler(QueuedMessage $queued): void { + $class = $queued->messageClass; + $handler = $this->handlerRegistry->getHandler($class); + if ($handler === null || !$this->metadataReader->isMessage($class)) { + throw new UnrecoverableMessageException('No handler registered for message ' . $class); + } + + try { + $data = json_decode($queued->body, true, flags: JSON_THROW_ON_ERROR); + if (!is_array($data)) { + throw new InvalidMessageException('Message body is not an object'); + } + + $message = $this->normalizer->denormalize($class, $data); + } catch (\JsonException|InvalidMessageException $e) { + throw new UnrecoverableMessageException('Could not decode message ' . $class . ': ' . $e->getMessage(), $e->getCode(), previous: $e); + } + + $userProperty = $this->metadataReader->get($class)->userProperty; + $user = null; + if ($userProperty !== null) { + /** @var mixed $userId */ + $userId = $message->$userProperty; + $user = is_string($userId) ? $this->userManager->get($userId) : null; + if (!$user instanceof IUser) { + throw new UnrecoverableMessageException('User ' . var_export($userId, true) . ' of message ' . $class . ' does not exist'); + } + + } + + try { + if ($user instanceof IUser) { + $this->userSession->setVolatileActiveUser($user); + $this->setupManager->setupForUser($user); + } + + $callable = [$this->container->get($handler['class']), $handler['method']]; + if (!is_callable($callable)) { + throw new UnrecoverableMessageException('Message handler ' . $handler['class'] . '::' . $handler['method'] . ' is not callable'); + } + + $callable($message); + } finally { + $this->setupManager->tearDown(); + $this->userSession->setVolatileActiveUser(null); + } + } + + /** + * @return int|null Seconds to wait before retrying, null if the message should not be retried + */ + private function getRetryDelay(QueuedMessage $queued, \Throwable $error): ?int { + if ($error instanceof UnrecoverableMessageException) { + return null; + } + + $metadata = $this->metadataReader->get($queued->messageClass); + if (!$error instanceof RecoverableMessageException && $queued->retryCount >= $metadata->maxRetries) { + return null; + } + + $delay = (float)$metadata->retryDelay * ($metadata->retryMultiplier ** (float)$queued->retryCount); + return (int)min((float)self::MAX_RETRY_DELAY, $delay); + } + + private function cleanUp(): void { + $this->tempManager->clean(); + if ($this->connection->inTransaction()) { + $this->connection->rollBack(); + $this->logger->warning('A message handler left a transaction open, it was rolled back'); + } + } +} diff --git a/lib/private/MessageQueue/HandlerRegistry.php b/lib/private/MessageQueue/HandlerRegistry.php new file mode 100644 index 0000000000000..4011603229664 --- /dev/null +++ b/lib/private/MessageQueue/HandlerRegistry.php @@ -0,0 +1,125 @@ + section of enabled apps. + * + * @psalm-type Handler = array{class: class-string, method: string} + * @psalm-suppress ClassMustBeFinal For unit tests + */ +class HandlerRegistry { + /** @var array|null */ + private ?array $handlers = null; + + private readonly ICache $cache; + + public function __construct( + private readonly IAppManager $appManager, + private readonly MessageMetadataReader $metadataReader, + private readonly LoggerInterface $logger, + ICacheFactory $cacheFactory, + ) { + $this->cache = $cacheFactory->createLocal('message-queue'); + } + + /** + * @return Handler|null + */ + public function getHandler(string $messageClass): ?array { + return $this->getHandlers()[$messageClass] ?? null; + } + + /** + * @return array + */ + public function getHandlers(): array { + if ($this->handlers !== null) { + return $this->handlers; + } + + $apps = []; + foreach ($this->appManager->getEnabledApps() as $app) { + $apps[$app] = $this->appManager->getAppVersion($app); + } + + $cacheKey = 'handlers-' . md5(json_encode($apps, JSON_THROW_ON_ERROR)); + + /** @var array|null $cached */ + $cached = $this->cache->get($cacheKey); + if (is_array($cached)) { + return $this->handlers = $cached; + } + + $handlers = []; + foreach (array_keys($apps) as $app) { + $info = $this->appManager->getAppInfo($app); + foreach ($info['message-handlers'] ?? [] as $handlerClass) { + $this->collect($app, $handlerClass, $handlers); + } + } + + $this->cache->set($cacheKey, $handlers, 3600); + return $this->handlers = $handlers; + } + + /** + * @param array $handlers + */ + private function collect(string $app, string $handlerClass, array &$handlers): void { + if (!class_exists($handlerClass)) { + $this->logger->error('Message handler ' . $handlerClass . ' of app ' . $app . ' does not exist'); + return; + } + + $class = new \ReflectionClass($handlerClass); + $methods = []; + if ($class->getAttributes(AsMessageHandler::class) !== []) { + if (!$class->hasMethod('__invoke')) { + $this->logger->error('Message handler ' . $handlerClass . ' of app ' . $app . ' is marked with #[AsMessageHandler] but has no __invoke method'); + return; + } + + $methods['__invoke'] = $class->getMethod('__invoke'); + } + + foreach ($class->getMethods(\ReflectionMethod::IS_PUBLIC) as $method) { + if ($method->getAttributes(AsMessageHandler::class) !== []) { + $methods[$method->getName()] = $method; + } + } + + if ($methods === []) { + $this->logger->error('Message handler ' . $handlerClass . ' of app ' . $app . ' has no #[AsMessageHandler] attribute'); + return; + } + + foreach ($methods as $method) { + $type = ($method->getParameters()[0] ?? null)?->getType(); + $messageClass = $type instanceof \ReflectionNamedType ? $type->getName() : null; + if ($messageClass === null || !$this->metadataReader->isMessage($messageClass)) { + $this->logger->error('First parameter of message handler ' . $handlerClass . '::' . $method->getName() . ' must be a class marked with #[AsMessage]'); + continue; + } + + if (isset($handlers[$messageClass])) { + $this->logger->error('Message ' . $messageClass . ' already has a handler, ignoring ' . $handlerClass . '::' . $method->getName()); + continue; + } + + $handlers[$messageClass] = ['class' => $handlerClass, 'method' => $method->getName()]; + } + } +} diff --git a/lib/private/MessageQueue/MessageBus.php b/lib/private/MessageQueue/MessageBus.php new file mode 100644 index 0000000000000..1ecdd7882d23d --- /dev/null +++ b/lib/private/MessageQueue/MessageBus.php @@ -0,0 +1,45 @@ +handlerRegistry->getHandler($class) === null) { + throw new InvalidMessageException('No handler registered for message ' . $class); + } + + $metadata = $this->metadataReader->get($class); + $data = $this->normalizer->normalize($message); + + $deduplicationHash = null; + if ($metadata->deduplicateBy !== []) { + $key = array_intersect_key($data, array_flip($metadata->deduplicateBy)); + ksort($key); + $deduplicationHash = hash('sha256', $class . json_encode($key, JSON_THROW_ON_ERROR)); + } + + $id = $this->store->add($metadata->queue, $class, json_encode($data, JSON_THROW_ON_ERROR), $deduplicationHash); + if ($id !== null) { + $this->afterRequestHandler->add($metadata->queue, $id); + } + } +} diff --git a/lib/private/MessageQueue/MessageMetadataReader.php b/lib/private/MessageQueue/MessageMetadataReader.php new file mode 100644 index 0000000000000..fd229bd4d33a6 --- /dev/null +++ b/lib/private/MessageQueue/MessageMetadataReader.php @@ -0,0 +1,48 @@ + */ + private array $cache = []; + + /** + * @psalm-assert-if-true class-string $class + */ + public function isMessage(string $class): bool { + return $this->read($class) instanceof AsMessage; + } + + /** + * @throws InvalidMessageException + */ + public function get(string $class): AsMessage { + $metadata = $this->read($class); + if (!$metadata instanceof AsMessage) { + throw new InvalidMessageException('Class ' . $class . ' is not marked with #[AsMessage]'); + } + + return $metadata; + } + + private function read(string $class): ?AsMessage { + if (!array_key_exists($class, $this->cache)) { + if (!class_exists($class)) { + return null; + } + + $attributes = (new \ReflectionClass($class))->getAttributes(AsMessage::class); + $this->cache[$class] = $attributes === [] ? null : $attributes[0]->newInstance(); + } + + return $this->cache[$class]; + } +} diff --git a/lib/private/MessageQueue/MessageNormalizer.php b/lib/private/MessageQueue/MessageNormalizer.php new file mode 100644 index 0000000000000..e7415c0082c89 --- /dev/null +++ b/lib/private/MessageQueue/MessageNormalizer.php @@ -0,0 +1,142 @@ + + * @throws InvalidMessageException + */ + public function normalize(object $message): array { + $class = $message::class; + $data = []; + foreach ($this->getConstructorParameters($class) as $parameter) { + $name = $parameter->getName(); + if (!property_exists($message, $name) || !(new \ReflectionProperty($message, $name))->isPublic()) { + throw new InvalidMessageException('Constructor parameter $' . $name . ' of ' . $class . ' must be a public promoted property'); + } + + $data[$name] = $this->normalizeValue($message->$name, $class . '::$' . $name); + } + + return $data; + } + + /** + * @template T of object + * @param class-string $class + * @param array $data + * @return T + * @throws InvalidMessageException + */ + public function denormalize(string $class, array $data): object { + $arguments = []; + foreach ($this->getConstructorParameters($class) as $parameter) { + $name = $parameter->getName(); + if (!array_key_exists($name, $data)) { + if ($parameter->isDefaultValueAvailable()) { + continue; + } + + throw new InvalidMessageException('Missing value for ' . $class . '::$' . $name); + } + + $arguments[$name] = $this->denormalizeValue($data[$name], $parameter->getType(), $class . '::$' . $name); + } + + return (new \ReflectionClass($class))->newInstanceArgs($arguments); + } + + /** + * @param class-string $class + * @return list<\ReflectionParameter> + */ + private function getConstructorParameters(string $class): array { + return (new \ReflectionClass($class))->getConstructor()?->getParameters() ?? []; + } + + private function normalizeValue(mixed $value, string $path, bool $inArray = false): int|float|string|bool|array|null { + if ($value === null || is_scalar($value)) { + return $value; + } + + if (is_array($value)) { + return array_map(fn (mixed $item): int|float|string|bool|array|null => $this->normalizeValue($item, $path, true), $value); + } + + if (!$inArray && is_object($value)) { + if ($value instanceof \BackedEnum) { + return $value->value; + } + + if ($value instanceof \DateTimeImmutable) { + return $value->format(self::DATE_FORMAT); + } + + if ($this->metadataReader->isMessage($value::class)) { + return $this->normalize($value); + } + } + + throw new InvalidMessageException('Unsupported value of type ' . get_debug_type($value) . ' in ' . $path); + } + + private function denormalizeValue(mixed $value, ?\ReflectionType $type, string $path): int|float|string|bool|array|object|null { + if (!$type instanceof \ReflectionNamedType) { + throw new InvalidMessageException('A single named type is required in ' . $path); + } + + if ($value === null) { + if ($type->allowsNull()) { + return null; + } + + throw new InvalidMessageException('Unexpected null value in ' . $path); + } + + $typeName = $type->getName(); + if ($type->isBuiltin()) { + return match (true) { + $typeName === 'int' && is_int($value) => $value, + $typeName === 'float' && (is_float($value) || is_int($value)) => (float)$value, + $typeName === 'string' && is_string($value) => $value, + $typeName === 'bool' && is_bool($value) => $value, + $typeName === 'array' && is_array($value) => $value, + default => throw new InvalidMessageException('Expected ' . $typeName . ' but got ' . get_debug_type($value) . ' in ' . $path), + }; + } + + if (is_subclass_of($typeName, \BackedEnum::class) && (is_int($value) || is_string($value))) { + return $typeName::tryFrom($value) ?? throw new InvalidMessageException('Invalid enum value in ' . $path); + } + + if (in_array($typeName, [\DateTimeImmutable::class, \DateTimeInterface::class], true) && is_string($value)) { + return \DateTimeImmutable::createFromFormat(self::DATE_FORMAT, $value) + ?: throw new InvalidMessageException('Invalid date in ' . $path); + } + + if (is_array($value) && $this->metadataReader->isMessage($typeName)) { + return $this->denormalize($typeName, $value); + } + + throw new InvalidMessageException('Unsupported type ' . $typeName . ' in ' . $path); + } +} diff --git a/lib/private/MessageQueue/MessageStore.php b/lib/private/MessageQueue/MessageStore.php new file mode 100644 index 0000000000000..86b3ccd5d3e8e --- /dev/null +++ b/lib/private/MessageQueue/MessageStore.php @@ -0,0 +1,223 @@ +hasPending($queue, $deduplicationHash)) { + return null; + } + + $now = $this->timeFactory->getTime(); + $id = $this->snowflakeGenerator->nextId(); + $qb = $this->connection->getQueryBuilder(); + $qb->insert(self::TABLE) + ->values([ + 'id' => $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT), + 'queue_name' => $qb->createNamedParameter($queue->value), + 'message_class' => $qb->createNamedParameter($messageClass), + 'body' => $qb->createNamedParameter($body), + 'created_at' => $qb->createNamedParameter($now, IQueryBuilder::PARAM_INT), + 'available_at' => $qb->createNamedParameter($now, IQueryBuilder::PARAM_INT), + 'deduplication_hash' => $qb->createNamedParameter($deduplicationHash), + ]) + ->executeStatement(); + return $id; + } + + /** + * Claims the next available message of the queue. + * + * SQLite doesn't support SKIP LOCKED and Oracle can't combine it with a row + * limit, so they claim optimistically and retry when another worker was faster. + */ + public function claimNext(Queue $queue): ?QueuedMessage { + $provider = $this->connection->getDatabaseProvider(); + if ($provider === IDBConnection::PLATFORM_SQLITE || $provider === IDBConnection::PLATFORM_ORACLE) { + return $this->claimNextOptimistically($queue); + } + + return $this->claimNextWithSkipLocked($queue); + } + + /** + * Locks the next available message, skipping the ones locked by other workers. + */ + private function claimNextWithSkipLocked(Queue $queue): ?QueuedMessage { + $this->connection->beginTransaction(); + try { + $message = $this->fetchAvailable($queue, skipLocked: true); + $claimed = $message instanceof QueuedMessage && $this->claim($message); + $this->connection->commit(); + } catch (\Throwable $throwable) { + $this->connection->rollBack(); + throw $throwable; + } + + return $claimed ? $message : null; + } + + private function claimNextOptimistically(Queue $queue): ?QueuedMessage { + for ($attempt = 0; $attempt < self::MAX_CLAIM_ATTEMPTS; ++$attempt) { + $message = $this->fetchAvailable($queue); + if (!$message instanceof QueuedMessage) { + return null; + } + + if ($this->claim($message)) { + return $message; + } + } + + return null; + } + + /** + * Claims a specific message, if it is available. + */ + public function claimById(string $id): ?QueuedMessage { + $message = $this->fetchAvailable(id: $id); + return $message instanceof QueuedMessage && $this->claim($message) ? $message : null; + } + + public function delete(string $id): void { + $qb = $this->connection->getQueryBuilder(); + $qb->delete(self::TABLE) + ->where($qb->expr()->eq('id', $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT))) + ->executeStatement(); + } + + /** + * Makes a claimed message available again after the given delay. + */ + public function retry(string $id, int $delay): void { + $qb = $this->connection->getQueryBuilder(); + $qb->update(self::TABLE) + ->set('retry_count', $qb->func()->add('retry_count', $qb->createNamedParameter(1, IQueryBuilder::PARAM_INT))) + ->set('available_at', $qb->createNamedParameter($this->timeFactory->getTime() + $delay, IQueryBuilder::PARAM_INT)) + ->set('delivered_at', $qb->createNamedParameter(null)) + ->where($qb->expr()->eq('id', $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT))) + ->executeStatement(); + } + + /** + * Moves a claimed message to the failed queue. + */ + public function fail(string $id, string $error): void { + $qb = $this->connection->getQueryBuilder(); + $qb->update(self::TABLE) + ->set('queue_name', $qb->createNamedParameter(self::FAILED_QUEUE)) + ->set('last_error', $qb->createNamedParameter($error)) + ->set('delivered_at', $qb->createNamedParameter(null)) + ->set('deduplication_hash', $qb->createNamedParameter(null)) + ->where($qb->expr()->eq('id', $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT))) + ->executeStatement(); + } + + private function fetchAvailable(?Queue $queue = null, ?string $id = null, bool $skipLocked = false): ?QueuedMessage { + $now = $this->timeFactory->getTime(); + $qb = $this->connection->getQueryBuilder(); + $qb->select('id', 'message_class', 'body', 'retry_count', 'delivered_at') + ->from(self::TABLE) + ->where($qb->expr()->lte('available_at', $qb->createNamedParameter($now, IQueryBuilder::PARAM_INT))) + ->andWhere($qb->expr()->orX( + $qb->expr()->isNull('delivered_at'), + $qb->expr()->lt('delivered_at', $qb->createNamedParameter($now - self::REDELIVER_TIMEOUT, IQueryBuilder::PARAM_INT)), + )) + ->orderBy('available_at', 'ASC') + ->addOrderBy('id', 'ASC') + ->setMaxResults(1); + if ($queue instanceof Queue) { + $qb->andWhere($qb->expr()->eq('queue_name', $qb->createNamedParameter($queue->value))); + } + + if ($id !== null) { + $qb->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT))); + } + + if ($skipLocked) { + $qb->forUpdate(ConflictResolutionMode::SkipLocked); + } + + $result = $qb->executeQuery(); + $row = $result->fetchAssociative(); + $result->closeCursor(); + if ($row === false) { + return null; + } + + return new QueuedMessage( + (string)$row['id'], + (string)$row['message_class'], + (string)$row['body'], + (int)$row['retry_count'], + $row['delivered_at'] === null ? null : (int)$row['delivered_at'], + ); + } + + /** + * Marks the message as delivered, unless another worker claimed it first. + */ + private function claim(QueuedMessage $message): bool { + $qb = $this->connection->getQueryBuilder(); + $qb->update(self::TABLE) + ->set('delivered_at', $qb->createNamedParameter($this->timeFactory->getTime(), IQueryBuilder::PARAM_INT)) + ->where($qb->expr()->eq('id', $qb->createNamedParameter($message->id, IQueryBuilder::PARAM_INT))); + if ($message->deliveredAt === null) { + $qb->andWhere($qb->expr()->isNull('delivered_at')); + } else { + $qb->andWhere($qb->expr()->eq('delivered_at', $qb->createNamedParameter($message->deliveredAt, IQueryBuilder::PARAM_INT))); + } + + return $qb->executeStatement() === 1; + } + + private function hasPending(Queue $queue, string $deduplicationHash): bool { + $qb = $this->connection->getQueryBuilder(); + $qb->select('id') + ->from(self::TABLE) + ->where($qb->expr()->eq('deduplication_hash', $qb->createNamedParameter($deduplicationHash))) + ->andWhere($qb->expr()->eq('queue_name', $qb->createNamedParameter($queue->value))) + ->andWhere($qb->expr()->isNull('delivered_at')) + ->setMaxResults(1); + $result = $qb->executeQuery(); + $found = $result->fetchOne() !== false; + $result->closeCursor(); + return $found; + } +} diff --git a/lib/private/MessageQueue/QueuedMessage.php b/lib/private/MessageQueue/QueuedMessage.php new file mode 100644 index 0000000000000..61d8f4c30cbae --- /dev/null +++ b/lib/private/MessageQueue/QueuedMessage.php @@ -0,0 +1,23 @@ +registerAlias(ILogFactory::class, LogFactory::class); $this->registerAlias(IJobList::class, JobList::class); + $this->registerAlias(IMessageBus::class, MessageBus::class); $this->registerService(Router::class, static function (Server $c) { $cacheFactory = $c->get(ICacheFactory::class); diff --git a/lib/public/App/AppInfoDefinition.php b/lib/public/App/AppInfoDefinition.php index 261057b2f770d..5db4218441cc4 100644 --- a/lib/public/App/AppInfoDefinition.php +++ b/lib/public/App/AppInfoDefinition.php @@ -248,6 +248,7 @@ * 'repair-steps'?: AppInfoFieldTypeRepairSteps, * 'two-factor-providers'?: list, * 'commands'?: list, + * 'message-handlers'?: list, * 'settings'?: AppInfoFieldTypeSettings, * 'activity'?: AppInfoFieldTypeActivity, * 'dashboard'?: AppInfoFieldTypeDashboard, diff --git a/lib/public/MessageQueue/Attribute/AsMessage.php b/lib/public/MessageQueue/Attribute/AsMessage.php new file mode 100644 index 0000000000000..3ebe7018af19e --- /dev/null +++ b/lib/public/MessageQueue/Attribute/AsMessage.php @@ -0,0 +1,60 @@ + $deduplicateBy Properties forming the deduplication key. While a message + * with the same key is pending, dispatching it again is a no-op. + * An empty list disables deduplication. + * @param string|null $userProperty Property holding a user id. The filesystem and user scope of + * that user are set up before the handler runs and torn down after. + * @since 36.0.0 + */ + public function __construct( + public Queue $queue = Queue::Default, + public int $maxRetries = 3, + public int $retryDelay = 60, + public float $retryMultiplier = 2.0, + public array $deduplicateBy = [], + public ?string $userProperty = null, + ) { + } +} diff --git a/lib/public/MessageQueue/Attribute/AsMessageHandler.php b/lib/public/MessageQueue/Attribute/AsMessageHandler.php new file mode 100644 index 0000000000000..50dc0541f983f --- /dev/null +++ b/lib/public/MessageQueue/Attribute/AsMessageHandler.php @@ -0,0 +1,55 @@ + and + * are resolved through the app container, so constructor injection works. + * + * Can be used either on an invokable class: + * + * ``` + * #[AsMessageHandler] + * final class GeneratePreviewHandler { + * public function __construct(private IPreview $preview) { + * } + * + * public function __invoke(GeneratePreview $message): void { + * // ... + * } + * } + * ``` + * + * Or on methods: + * + * ``` + * final class FileTasks { + * #[AsMessageHandler] + * public function preview(GeneratePreview $message): void { + * } + * + * #[AsMessageHandler] + * public function scan(ScanFolder $message): void { + * } + * } + * ``` + * + * Throw an UnrecoverableMessageException to fail the message without retrying. + * Any other exception triggers a retry according to the message's retry settings. + * + * @since 36.0.0 + */ +#[\Attribute(\Attribute::TARGET_CLASS | \Attribute::TARGET_METHOD)] +#[Consumable(since: '36.0.0')] +final class AsMessageHandler { +} diff --git a/lib/public/MessageQueue/Exception/InvalidMessageException.php b/lib/public/MessageQueue/Exception/InvalidMessageException.php new file mode 100644 index 0000000000000..7955503a493fd --- /dev/null +++ b/lib/public/MessageQueue/Exception/InvalidMessageException.php @@ -0,0 +1,19 @@ + + + + + diff --git a/resources/app-info-shipped.xsd b/resources/app-info-shipped.xsd index c7f3808ca9dd9..7f910d7b0b354 100644 --- a/resources/app-info-shipped.xsd +++ b/resources/app-info-shipped.xsd @@ -55,6 +55,8 @@ minOccurs="0" maxOccurs="1"/> + + + + + + + + + + + + + + createMock(IConfig::class); + $config->method('getSystemValueBool')->with('message_queue.handle_after_request', true)->willReturn($enabled); + + return new TestAfterRequestHandler( + $consumer, + $config, + $this->createMock(ISession::class), + $this->createMock(IUserSession::class), + $this->createMock(ISetupManager::class), + $lockingProvider ?? $this->createMock(ILockingProvider::class), + $logger ?? $this->createMock(LoggerInterface::class), + $isCLI, + ); + } + + public function testEnabled(): void { + $consumer = $this->createMock(Consumer::class); + $consumer->expects($this->once())->method('consumeMessages')->with(['1', '2', '3'], 10, 5); + + $handler = $this->createHandler($consumer); + $handler->add(Queue::High, '1'); + $handler->add(Queue::Default, '2'); + $handler->add(Queue::High, '3'); + $handler->add(Queue::Low, '4'); + $this->assertCount(1, $handler->shutdownFunctions); + + $handler->run(); + $handler->run(); + $this->assertTrue($handler->requestFinished); + } + + public function testDisabledOnCli(): void { + $consumer = $this->createMock(Consumer::class); + $consumer->expects($this->never())->method('consumeMessages'); + + $handler = $this->createHandler($consumer, isCLI: true); + $handler->add(Queue::High, '1'); + $handler->run(); + $this->assertSame([], $handler->shutdownFunctions); + } + + public function testDisabledByConfig(): void { + $consumer = $this->createMock(Consumer::class); + $consumer->expects($this->never())->method('consumeMessages'); + + $handler = $this->createHandler($consumer, enabled: false); + $handler->add(Queue::High, '1'); + $handler->run(); + $this->assertSame([], $handler->shutdownFunctions); + } + + public function testErrorsAreLoggedAndLocksReleased(): void { + $consumer = $this->createMock(Consumer::class); + $consumer->method('consumeMessages')->willThrowException(new \RuntimeException('Broken')); + $logger = $this->createMock(LoggerInterface::class); + $logger->expects($this->once())->method('error'); + $lockingProvider = $this->createMock(ILockingProvider::class); + $lockingProvider->expects($this->once())->method('releaseAll'); + + $handler = $this->createHandler($consumer, lockingProvider: $lockingProvider, logger: $logger); + $handler->add(Queue::High, '1'); + $handler->run(); + } +} diff --git a/tests/lib/MessageQueue/Fixtures/DuplicateHandler.php b/tests/lib/MessageQueue/Fixtures/DuplicateHandler.php new file mode 100644 index 0000000000000..d21520e8e2bed --- /dev/null +++ b/tests/lib/MessageQueue/Fixtures/DuplicateHandler.php @@ -0,0 +1,20 @@ + */ + public array $handled = []; + + public function __invoke(TestMessage $message): void { + $this->handled[] = $message; + } +} diff --git a/tests/lib/MessageQueue/Fixtures/LowMessage.php b/tests/lib/MessageQueue/Fixtures/LowMessage.php new file mode 100644 index 0000000000000..02c17a9deb2fe --- /dev/null +++ b/tests/lib/MessageQueue/Fixtures/LowMessage.php @@ -0,0 +1,19 @@ + */ + public array $users = []; + + /** @var list */ + public array $low = []; + + public function __construct( + private readonly IUserSession $userSession, + ) { + } + + #[AsMessageHandler] + public function user(UserMessage $message): void { + $this->users[] = $this->userSession->getUser()?->getUID(); + } + + #[AsMessageHandler] + public function fail(FailingMessage $message): void { + if ($message->unrecoverable) { + throw new UnrecoverableMessageException('Permanent failure'); + } + + if ($message->recoverable) { + throw new RecoverableMessageException('Service unavailable'); + } + + throw new \RuntimeException('Temporary failure'); + } + + #[AsMessageHandler] + public function low(LowMessage $message): void { + $this->low[] = $message; + } + + public function notAHandler(TestMessage $message): void { + } +} diff --git a/tests/lib/MessageQueue/Fixtures/NestedMessage.php b/tests/lib/MessageQueue/Fixtures/NestedMessage.php new file mode 100644 index 0000000000000..44f89e63d1626 --- /dev/null +++ b/tests/lib/MessageQueue/Fixtures/NestedMessage.php @@ -0,0 +1,18 @@ + */ + public array $shutdownFunctions = []; + + public bool $requestFinished = false; + + #[\Override] + protected function canFinishRequest(): bool { + return true; + } + + #[\Override] + protected function registerShutdownFunction(callable $callback): void { + $this->shutdownFunctions[] = $callback; + } + + #[\Override] + protected function finishRequest(): void { + $this->requestFinished = true; + } +} diff --git a/tests/lib/MessageQueue/Fixtures/TestMessage.php b/tests/lib/MessageQueue/Fixtures/TestMessage.php new file mode 100644 index 0000000000000..39820b9d1be40 --- /dev/null +++ b/tests/lib/MessageQueue/Fixtures/TestMessage.php @@ -0,0 +1,25 @@ +appManager = $this->createMock(IAppManager::class); + $this->logger = $this->createMock(LoggerInterface::class); + $this->cache = new ArrayCache(); + $cacheFactory = $this->createMock(ICacheFactory::class); + $cacheFactory->method('createLocal')->willReturn($this->cache); + + $this->appManager->method('getEnabledApps')->willReturn(['app1', 'app2']); + $this->appManager->method('getAppVersion')->willReturn('1.0.0'); + $this->appManager->method('getAppInfo')->willReturnMap([ + ['app1', false, null, ['message-handlers' => [InvokableHandler::class, MultiHandler::class]]], + ['app2', false, null, ['message-handlers' => ['handler' => DuplicateHandler::class]]], + ]); + + $this->registry = new HandlerRegistry($this->appManager, new MessageMetadataReader(), $this->logger, $cacheFactory); + } + + public function testCollectsHandlers(): void { + $this->logger->expects($this->exactly(2))->method('error'); + + $this->assertSame([ + TestMessage::class => ['class' => InvokableHandler::class, 'method' => '__invoke'], + UserMessage::class => ['class' => MultiHandler::class, 'method' => 'user'], + FailingMessage::class => ['class' => MultiHandler::class, 'method' => 'fail'], + LowMessage::class => ['class' => MultiHandler::class, 'method' => 'low'], + ], $this->registry->getHandlers()); + $this->assertNull($this->registry->getHandler(NotAMessage::class)); + } + + public function testUsesCachedHandlers(): void { + $cached = [TestMessage::class => ['class' => InvokableHandler::class, 'method' => '__invoke']]; + $this->cache->set('handlers-' . md5(json_encode(['app1' => '1.0.0', 'app2' => '1.0.0'], JSON_THROW_ON_ERROR)), $cached); + $this->appManager->expects($this->never())->method('getAppInfo'); + + $this->assertSame($cached, $this->registry->getHandlers()); + } + + public function testStoresHandlersInCache(): void { + $handlers = $this->registry->getHandlers(); + $this->assertSame($handlers, $this->cache->get('handlers-' . md5(json_encode(['app1' => '1.0.0', 'app2' => '1.0.0'], JSON_THROW_ON_ERROR)))); + } +} diff --git a/tests/lib/MessageQueue/MessageNormalizerTest.php b/tests/lib/MessageQueue/MessageNormalizerTest.php new file mode 100644 index 0000000000000..c7a7f347df9c8 --- /dev/null +++ b/tests/lib/MessageQueue/MessageNormalizerTest.php @@ -0,0 +1,84 @@ +normalizer = new MessageNormalizer(new MessageMetadataReader()); + } + + public function testRoundTrip(): void { + $message = new TestMessage( + id: 42, + text: 'hello', + status: Status::Disabled, + date: new \DateTimeImmutable('2026-10-04 12:34:56.123456+02:00'), + nested: new NestedMessage('child'), + list: ['a' => [1, 2], 'b' => null], + ratio: 2.5, + ); + + $data = $this->normalizer->normalize($message); + $this->assertSame([ + 'id' => 42, + 'text' => 'hello', + 'status' => 'disabled', + 'date' => '2026-10-04T12:34:56.123456+02:00', + 'nested' => ['name' => 'child'], + 'list' => ['a' => [1, 2], 'b' => null], + 'ratio' => 2.5, + ], $data); + + $json = json_decode(json_encode($data, JSON_THROW_ON_ERROR), true, flags: JSON_THROW_ON_ERROR); + $this->assertIsArray($json); + $this->assertEquals($message, $this->normalizer->denormalize(TestMessage::class, $json)); + } + + public function testDenormalizeUsesDefaultsForMissingValues(): void { + $message = $this->normalizer->denormalize(TestMessage::class, ['id' => 1, 'ratio' => 3]); + $this->assertEquals(new TestMessage(id: 1, ratio: 3.0), $message); + } + + public function testNormalizeRejectsObjectsInArrays(): void { + $this->expectException(InvalidMessageException::class); + $this->normalizer->normalize(new TestMessage(id: 1, list: [Status::Active])); + } + + /** + * @return array}> + */ + public static function invalidDataProvider(): array { + return [ + 'missing required value' => [['text' => 'a']], + 'wrong scalar type' => [['id' => '1']], + 'unexpected null' => [['id' => null]], + 'invalid enum value' => [['id' => 1, 'status' => 'unknown']], + 'invalid date' => [['id' => 1, 'date' => 'yesterday']], + 'invalid nested message' => [['id' => 1, 'nested' => 'child']], + ]; + } + + #[DataProvider('invalidDataProvider')] + public function testDenormalizeRejectsInvalidData(array $data): void { + $this->expectException(InvalidMessageException::class); + $this->normalizer->denormalize(TestMessage::class, $data); + } +} diff --git a/tests/lib/MessageQueue/MessageQueueTest.php b/tests/lib/MessageQueue/MessageQueueTest.php new file mode 100644 index 0000000000000..ccdf9f8453b39 --- /dev/null +++ b/tests/lib/MessageQueue/MessageQueueTest.php @@ -0,0 +1,342 @@ +connection = Server::get(IDBConnection::class); + $this->clearTable(); + + $timeFactory = $this->createMock(ITimeFactory::class); + $timeFactory->method('getTime')->willReturnCallback(fn (): int => $this->now); + + $userSession = $this->createMock(IUserSession::class); + $userSession->method('setVolatileActiveUser')->willReturnCallback(function (?IUser $user): void { + $this->activeUser = $user; + }); + $userSession->method('getUser')->willReturnCallback(fn (): ?IUser => $this->activeUser); + + $alice = $this->createMock(IUser::class); + $alice->method('getUID')->willReturn('alice'); + $userManager = $this->createMock(IUserManager::class); + $userManager->method('get')->willReturnCallback(static fn (string $uid) => $uid === 'alice' ? $alice : null); + + $this->setupManager = $this->createMock(ISetupManager::class); + $this->invokableHandler = new InvokableHandler(); + $this->multiHandler = new MultiHandler($userSession); + + $container = $this->createMock(ContainerInterface::class); + $container->method('get')->willReturnMap([ + [InvokableHandler::class, $this->invokableHandler], + [MultiHandler::class, $this->multiHandler], + ]); + + $handlerRegistry = $this->createMock(HandlerRegistry::class); + $handlerRegistry->method('getHandler')->willReturnCallback(static fn (string $class): ?array => match ($class) { + TestMessage::class => ['class' => InvokableHandler::class, 'method' => '__invoke'], + UserMessage::class => ['class' => MultiHandler::class, 'method' => 'user'], + FailingMessage::class => ['class' => MultiHandler::class, 'method' => 'fail'], + LowMessage::class => ['class' => MultiHandler::class, 'method' => 'low'], + default => null, + }); + + $logger = new NullLogger(); + $metadataReader = new MessageMetadataReader(); + $normalizer = new MessageNormalizer($metadataReader); + $store = new MessageStore($this->connection, $timeFactory, Server::get(ISnowflakeGenerator::class)); + $this->consumer = new Consumer( + $store, + $handlerRegistry, + $metadataReader, + $normalizer, + $container, + $userManager, + $userSession, + $this->setupManager, + $this->connection, + $this->createMock(ITempManager::class), + $logger, + ); + $this->session = $this->createMock(ISession::class); + $this->lockingProvider = $this->createMock(ILockingProvider::class); + $config = $this->createMock(IConfig::class); + $config->method('getSystemValueBool')->willReturn(true); + $this->afterRequestHandler = new TestAfterRequestHandler($this->consumer, $config, $this->session, $userSession, $this->setupManager, $this->lockingProvider, $logger, false); + $this->bus = new MessageBus($store, $handlerRegistry, $metadataReader, $normalizer, $this->afterRequestHandler); + } + + #[\Override] + protected function tearDown(): void { + $this->clearTable(); + parent::tearDown(); + } + + private function clearTable(): void { + $this->connection->getQueryBuilder()->delete(MessageStore::TABLE)->executeStatement(); + } + + /** + * @return list + */ + private function getRows(): array { + $qb = $this->connection->getQueryBuilder(); + $result = $qb->select('queue_name', 'available_at', 'retry_count', 'last_error') + ->from(MessageStore::TABLE) + ->orderBy('id') + ->executeQuery(); + $rows = array_map(static fn (array $row): array => [ + 'queue_name' => (string)$row['queue_name'], + 'available_at' => (int)$row['available_at'], + 'retry_count' => (int)$row['retry_count'], + 'last_error' => $row['last_error'] === null ? null : (string)$row['last_error'], + ], $result->fetchAllAssociative()); + $result->closeCursor(); + return $rows; + } + + private function consume(): void { + $this->consumer->consume(stopWhenEmpty: true); + } + + public function testDispatchAndConsume(): void { + $message = new TestMessage(id: 1, text: 'hello'); + $this->bus->dispatch($message); + + $this->assertSame('high', $this->getRows()[0]['queue_name']); + $this->assertSame([], $this->invokableHandler->handled); + + $this->consume(); + + $this->assertEquals([$message], $this->invokableHandler->handled); + $this->assertSame([], $this->getRows()); + } + + public function testDispatchRejectsMessageWithoutHandler(): void { + $this->expectException(InvalidMessageException::class); + $this->bus->dispatch(new NotAMessage(1)); + } + + public function testDispatchRejectsUnsupportedValues(): void { + $this->expectException(InvalidMessageException::class); + $this->bus->dispatch(new TestMessage(id: 1, list: [new \stdClass()])); + } + + public function testDeduplication(): void { + $this->bus->dispatch(new TestMessage(id: 1, text: 'first')); + $this->bus->dispatch(new TestMessage(id: 1, text: 'second')); + $this->bus->dispatch(new TestMessage(id: 2)); + $this->assertCount(2, $this->getRows()); + + $this->consume(); + $this->assertSame(['first', 'default'], array_map(static fn (TestMessage $m): string => $m->text, $this->invokableHandler->handled)); + + $this->bus->dispatch(new TestMessage(id: 1, text: 'third')); + $this->assertCount(1, $this->getRows()); + } + + public function testUserScope(): void { + $this->setupManager->expects($this->once())->method('setupForUser'); + $this->setupManager->expects($this->once())->method('tearDown'); + + $this->bus->dispatch(new UserMessage('alice')); + $this->consume(); + + $this->assertSame(['alice'], $this->multiHandler->users); + $this->assertNull($this->activeUser); + } + + public function testUnknownUserFailsWithoutRetry(): void { + $this->bus->dispatch(new UserMessage('bob')); + $this->consume(); + + $this->assertSame([], $this->multiHandler->users); + $rows = $this->getRows(); + $this->assertCount(1, $rows); + $this->assertSame('failed', $rows[0]['queue_name']); + } + + public function testRetryThenFailureQueue(): void { + $this->bus->dispatch(new FailingMessage()); + $this->consume(); + + $rows = $this->getRows(); + $this->assertCount(1, $rows); + $this->assertSame('default', $rows[0]['queue_name']); + $this->assertSame(1, $rows[0]['retry_count']); + $this->assertSame($this->now + 30, $rows[0]['available_at']); + + $this->now += 30; + $this->consume(); + + $rows = $this->getRows(); + $this->assertCount(1, $rows); + $this->assertSame('failed', $rows[0]['queue_name']); + $this->assertSame('RuntimeException: Temporary failure', $rows[0]['last_error']); + } + + public function testRecoverableIsRetriedWithBackoff(): void { + $this->bus->dispatch(new FailingMessage(recoverable: true)); + + foreach ([30, 60, 120] as $retry => $delay) { + $this->consume(); + $rows = $this->getRows(); + $this->assertSame('default', $rows[0]['queue_name']); + $this->assertSame($retry + 1, $rows[0]['retry_count']); + $this->assertSame($this->now + $delay, $rows[0]['available_at']); + $this->now += $delay; + } + } + + public function testUnrecoverableGoesToFailureQueue(): void { + $this->bus->dispatch(new FailingMessage(unrecoverable: true)); + $this->consume(); + + $rows = $this->getRows(); + $this->assertCount(1, $rows); + $this->assertSame('failed', $rows[0]['queue_name']); + $this->assertSame(UnrecoverableMessageException::class . ': Permanent failure', $rows[0]['last_error']); + } + + public function testUndecodableMessageIsMovedToFailureQueue(): void { + $qb = $this->connection->getQueryBuilder(); + $qb->insert(MessageStore::TABLE) + ->values([ + 'id' => $qb->createNamedParameter(Server::get(ISnowflakeGenerator::class)->nextId(), IQueryBuilder::PARAM_INT), + 'queue_name' => $qb->createNamedParameter('default'), + 'message_class' => $qb->createNamedParameter(NotAMessage::class), + 'body' => $qb->createNamedParameter('{"id":1}'), + 'created_at' => $qb->createNamedParameter($this->now, IQueryBuilder::PARAM_INT), + 'available_at' => $qb->createNamedParameter($this->now, IQueryBuilder::PARAM_INT), + ]) + ->executeStatement(); + + $this->consume(); + + $rows = $this->getRows(); + $this->assertCount(1, $rows); + $this->assertSame('failed', $rows[0]['queue_name']); + } + + public function testHandleAfterRequest(): void { + $this->session->expects($this->once())->method('close'); + $this->lockingProvider->expects($this->once())->method('releaseAll'); + + $this->bus->dispatch(new TestMessage(1)); + $this->bus->dispatch(new UserMessage('alice')); + $this->assertCount(1, $this->afterRequestHandler->shutdownFunctions); + + $this->afterRequestHandler->run(); + + $this->assertTrue($this->afterRequestHandler->requestFinished); + $this->assertCount(1, $this->invokableHandler->handled); + $this->assertSame(['alice'], $this->multiHandler->users); + $this->assertSame([], $this->getRows()); + } + + public function testHandleAfterRequestSkipsLowMessages(): void { + $this->bus->dispatch(new LowMessage(1)); + $this->assertSame([], $this->afterRequestHandler->shutdownFunctions); + + $this->afterRequestHandler->run(); + + $this->assertFalse($this->afterRequestHandler->requestFinished); + $this->assertCount(1, $this->getRows()); + } + + public function testHandleAfterRequestRespectsMessageLimit(): void { + for ($i = 1; $i <= 7; ++$i) { + $this->bus->dispatch(new TestMessage($i)); + } + + $this->afterRequestHandler->run(); + + $this->assertSame([1, 2, 3, 4, 5], array_map(static fn (TestMessage $m): int => $m->id, $this->invokableHandler->handled)); + $this->assertCount(2, $this->getRows()); + } + + public function testHandleAfterRequestOnlyHandlesOwnMessages(): void { + $this->bus->dispatch(new TestMessage(1)); + $this->consume(); + $this->bus->dispatch(new TestMessage(2)); + + $this->afterRequestHandler->run(); + + $this->assertSame([1, 2], array_map(static fn (TestMessage $m): int => $m->id, $this->invokableHandler->handled)); + } + + public function testHandleAfterRequestRetriesFailures(): void { + $this->bus->dispatch(new FailingMessage()); + + $this->afterRequestHandler->run(); + + $rows = $this->getRows(); + $this->assertCount(1, $rows); + $this->assertSame('default', $rows[0]['queue_name']); + $this->assertSame(1, $rows[0]['retry_count']); + } +} diff --git a/version.php b/version.php index 31e5cb2bcda4c..ca54c28709bc6 100644 --- a/version.php +++ b/version.php @@ -11,7 +11,7 @@ // between betas, final and RCs. This is _not_ the public version number. Reset minor/patch level // when updating major/minor version number. -$OC_Version = [36, 0, 0, 0]; +$OC_Version = [36, 0, 0, 1]; // The human-readable string $OC_VersionString = '36.0.0 dev';