diff --git a/core/Command/TaskProcessing/WorkerCommand.php b/core/Command/TaskProcessing/WorkerCommand.php new file mode 100644 index 0000000000000..6f419daab3ed4 --- /dev/null +++ b/core/Command/TaskProcessing/WorkerCommand.php @@ -0,0 +1,182 @@ +setName('taskprocessing:worker') + ->setDescription('Run a dedicated worker for synchronous TaskProcessing providers') + ->addOption( + 'timeout', + 't', + InputOption::VALUE_OPTIONAL, + 'Duration in seconds after which the worker exits (0 = run indefinitely)', + 0 + ) + ->addOption( + 'interval', + 'i', + InputOption::VALUE_OPTIONAL, + 'Sleep duration in seconds between polling iterations when no task was processed', + 1 + ) + ->addOption( + 'once', + null, + InputOption::VALUE_NONE, + 'Process at most one task then exit' + ) + ->addOption( + 'taskTypes', + null, + InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, + 'Only process tasks of the given task type IDs (can be specified multiple times)' + ); + parent::configure(); + } + + protected function execute(InputInterface $input, OutputInterface $output): int { + $startTime = time(); + $timeout = (int)$input->getOption('timeout'); + $interval = (int)$input->getOption('interval'); + $once = $input->getOption('once') === true; + /** @var list $taskTypes */ + $taskTypes = $input->getOption('taskTypes'); + + if ($timeout > 0) { + $output->writeln('Task processing worker will stop after ' . $timeout . ' seconds'); + } + + while (true) { + // Stop if timeout exceeded + if ($timeout > 0 && ($startTime + $timeout) < time()) { + $output->writeln('Timeout reached, exiting...', OutputInterface::VERBOSITY_VERBOSE); + break; + } + + // Handle SIGTERM/SIGINT gracefully + try { + $this->abortIfInterrupted(); + } catch (InterruptedException $e) { + $output->writeln('Task processing worker stopped'); + break; + } + + $processedTask = $this->processNextTask($output, $taskTypes); + + if ($once) { + break; + } + + if (!$processedTask) { + $output->writeln('No task processed, waiting ' . $interval . ' second(s)...', OutputInterface::VERBOSITY_VERBOSE); + sleep($interval); + } + } + + return 0; + } + + /** + * Attempt to process one task across all preferred synchronous providers. + * + * To avoid starvation, all eligible task types are first collected and then + * the oldest scheduled task across all of them is fetched in a single query. + * This ensures that tasks are processed in the order they were scheduled, + * regardless of which provider handles them. + * + * @param list $taskTypes When non-empty, only providers for these task type IDs are considered. + * @return bool True if a task was processed, false if no task was found + */ + private function processNextTask(OutputInterface $output, array $taskTypes = []): bool { + $providers = $this->taskProcessingManager->getProviders(); + + // Build a map of eligible taskTypeId => provider for all preferred synchronous providers + /** @var array $eligibleProviders */ + $eligibleProviders = []; + foreach ($providers as $provider) { + if (!$provider instanceof ISynchronousProvider) { + continue; + } + + $taskTypeId = $provider->getTaskTypeId(); + + // If a task type whitelist was provided, skip providers not in the list + if (!empty($taskTypes) && !in_array($taskTypeId, $taskTypes, true)) { + continue; + } + + // Only use this provider if it is the preferred one for the task type + try { + $preferredProvider = $this->taskProcessingManager->getPreferredProvider($taskTypeId); + } catch (Exception $e) { + $this->logger->error('Failed to get preferred provider for task type ' . $taskTypeId, ['exception' => $e]); + continue; + } + + if ($provider->getId() !== $preferredProvider->getId()) { + continue; + } + + $eligibleProviders[$taskTypeId] = $provider; + } + + if (empty($eligibleProviders)) { + return false; + } + + // Fetch the oldest scheduled task across all eligible task types in one query. + // This naturally prevents starvation: regardless of how many tasks one provider + // has queued, another provider's older tasks will be picked up first. + try { + $task = $this->taskProcessingManager->getNextScheduledTask(array_keys($eligibleProviders)); + } catch (NotFoundException) { + return false; + } catch (Exception $e) { + $this->logger->error('Unknown error while retrieving scheduled TaskProcessing tasks', ['exception' => $e]); + return false; + } + + $taskTypeId = $task->getTaskTypeId(); + $provider = $eligibleProviders[$taskTypeId]; + + $output->writeln( + 'Processing task ' . $task->getId() . ' of type ' . $taskTypeId . ' with provider ' . $provider->getId(), + OutputInterface::VERBOSITY_VERBOSE + ); + + $this->taskProcessingManager->processTask($task, $provider); + + $output->writeln( + 'Finished processing task ' . $task->getId(), + OutputInterface::VERBOSITY_VERBOSE + ); + + return true; + } +} diff --git a/core/register_command.php b/core/register_command.php index 58aed05ba68a6..d28c1633c62bb 100644 --- a/core/register_command.php +++ b/core/register_command.php @@ -91,6 +91,7 @@ use OC\Core\Command\TaskProcessing\EnabledCommand; use OC\Core\Command\TaskProcessing\GetCommand; use OC\Core\Command\TaskProcessing\Statistics; +use OC\Core\Command\TaskProcessing\WorkerCommand; use OC\Core\Command\TwoFactorAuth\Cleanup; use OC\Core\Command\TwoFactorAuth\Enforce; use OC\Core\Command\TwoFactorAuth\State; @@ -255,6 +256,7 @@ $application->add(Server::get(Command\TaskProcessing\ListCommand::class)); $application->add(Server::get(Statistics::class)); $application->add(Server::get(Command\TaskProcessing\Cleanup::class)); + $application->add(Server::get(WorkerCommand::class)); $application->add(Server::get(RedisCommand::class)); $application->add(Server::get(DistributedClear::class)); diff --git a/lib/composer/composer/autoload_classmap.php b/lib/composer/composer/autoload_classmap.php index 93f5def20a52a..a7de4feb7aa33 100644 --- a/lib/composer/composer/autoload_classmap.php +++ b/lib/composer/composer/autoload_classmap.php @@ -1422,6 +1422,7 @@ 'OC\\Core\\Command\\TaskProcessing\\GetCommand' => $baseDir . '/core/Command/TaskProcessing/GetCommand.php', 'OC\\Core\\Command\\TaskProcessing\\ListCommand' => $baseDir . '/core/Command/TaskProcessing/ListCommand.php', 'OC\\Core\\Command\\TaskProcessing\\Statistics' => $baseDir . '/core/Command/TaskProcessing/Statistics.php', + 'OC\\Core\\Command\\TaskProcessing\\WorkerCommand' => $baseDir . '/core/Command/TaskProcessing/WorkerCommand.php', 'OC\\Core\\Command\\TwoFactorAuth\\Base' => $baseDir . '/core/Command/TwoFactorAuth/Base.php', 'OC\\Core\\Command\\TwoFactorAuth\\Cleanup' => $baseDir . '/core/Command/TwoFactorAuth/Cleanup.php', 'OC\\Core\\Command\\TwoFactorAuth\\Disable' => $baseDir . '/core/Command/TwoFactorAuth/Disable.php', diff --git a/lib/composer/composer/autoload_static.php b/lib/composer/composer/autoload_static.php index bc27a12cf46b6..0c99e77ef1eaa 100644 --- a/lib/composer/composer/autoload_static.php +++ b/lib/composer/composer/autoload_static.php @@ -1463,6 +1463,7 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2 'OC\\Core\\Command\\TaskProcessing\\GetCommand' => __DIR__ . '/../../..' . '/core/Command/TaskProcessing/GetCommand.php', 'OC\\Core\\Command\\TaskProcessing\\ListCommand' => __DIR__ . '/../../..' . '/core/Command/TaskProcessing/ListCommand.php', 'OC\\Core\\Command\\TaskProcessing\\Statistics' => __DIR__ . '/../../..' . '/core/Command/TaskProcessing/Statistics.php', + 'OC\\Core\\Command\\TaskProcessing\\WorkerCommand' => __DIR__ . '/../../..' . '/core/Command/TaskProcessing/WorkerCommand.php', 'OC\\Core\\Command\\TwoFactorAuth\\Base' => __DIR__ . '/../../..' . '/core/Command/TwoFactorAuth/Base.php', 'OC\\Core\\Command\\TwoFactorAuth\\Cleanup' => __DIR__ . '/../../..' . '/core/Command/TwoFactorAuth/Cleanup.php', 'OC\\Core\\Command\\TwoFactorAuth\\Disable' => __DIR__ . '/../../..' . '/core/Command/TwoFactorAuth/Disable.php', diff --git a/tests/Core/Command/TaskProcessing/WorkerCommandTest.php b/tests/Core/Command/TaskProcessing/WorkerCommandTest.php new file mode 100644 index 0000000000000..8d516460aad93 --- /dev/null +++ b/tests/Core/Command/TaskProcessing/WorkerCommandTest.php @@ -0,0 +1,355 @@ +manager = $this->createMock(IManager::class); + $this->logger = $this->createMock(LoggerInterface::class); + $this->command = new WorkerCommand($this->manager, $this->logger); + } + + /** + * Helper to create a minimal ISynchronousProvider mock. + */ + private function createProvider(string $id, string $taskTypeId): ISynchronousProvider&MockObject { + $provider = $this->createMock(ISynchronousProvider::class); + $provider->method('getId')->willReturn($id); + $provider->method('getTaskTypeId')->willReturn($taskTypeId); + return $provider; + } + + /** + * Helper to create a real Task with a given id. + */ + private function createTask(int $id): Task { + $task = new Task('test_task_type', [], 'testapp', null); + $task->setId($id); + return $task; + } + + public function testOnceExitsAfterNoTask(): void { + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([]); + + $input = new ArrayInput(['--once' => true], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } + + public function testOnceProcessesOneTask(): void { + $taskTypeId = 'test_task_type'; + $provider = $this->createProvider('test_provider', $taskTypeId); + $task = $this->createTask(42); + + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([$provider]); + + $this->manager->expects($this->once()) + ->method('getPreferredProvider') + ->with($taskTypeId) + ->willReturn($provider); + + $this->manager->expects($this->once()) + ->method('getNextScheduledTask') + ->with([$taskTypeId]) + ->willReturn($task); + + $this->manager->expects($this->once()) + ->method('processTask') + ->with($task, $provider) + ->willReturn(true); + + $input = new ArrayInput(['--once' => true], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } + + public function testSkipsNonSynchronousProviders(): void { + // A provider that is NOT an ISynchronousProvider + $nonSyncProvider = $this->createMock(\OCP\TaskProcessing\IProvider::class); + $nonSyncProvider->method('getId')->willReturn('non_sync_provider'); + $nonSyncProvider->method('getTaskTypeId')->willReturn('some_type'); + + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([$nonSyncProvider]); + + $this->manager->expects($this->never()) + ->method('getPreferredProvider'); + + $this->manager->expects($this->never()) + ->method('getNextScheduledTask'); + + $input = new ArrayInput(['--once' => true], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } + + public function testSkipsNonPreferredProviders(): void { + $taskTypeId = 'test_task_type'; + $provider = $this->createProvider('provider_a', $taskTypeId); + $preferredProvider = $this->createProvider('provider_b', $taskTypeId); + + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([$provider]); + + $this->manager->expects($this->once()) + ->method('getPreferredProvider') + ->with($taskTypeId) + ->willReturn($preferredProvider); + + // provider_a is not preferred (provider_b is), so getNextScheduledTask is never called + $this->manager->expects($this->never()) + ->method('getNextScheduledTask'); + + $input = new ArrayInput(['--once' => true], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } + + public function testContinuesWhenNoTaskFound(): void { + $taskTypeId = 'test_task_type'; + $provider = $this->createProvider('test_provider', $taskTypeId); + + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([$provider]); + + $this->manager->expects($this->once()) + ->method('getPreferredProvider') + ->with($taskTypeId) + ->willReturn($provider); + + $this->manager->expects($this->once()) + ->method('getNextScheduledTask') + ->with([$taskTypeId]) + ->willThrowException(new NotFoundException()); + + $this->manager->expects($this->never()) + ->method('processTask'); + + $input = new ArrayInput(['--once' => true], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } + + public function testLogsErrorAndContinuesOnException(): void { + $taskTypeId = 'test_task_type'; + $provider = $this->createProvider('test_provider', $taskTypeId); + + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([$provider]); + + $this->manager->expects($this->once()) + ->method('getPreferredProvider') + ->with($taskTypeId) + ->willReturn($provider); + + $exception = new Exception('DB error'); + $this->manager->expects($this->once()) + ->method('getNextScheduledTask') + ->with([$taskTypeId]) + ->willThrowException($exception); + + $this->logger->expects($this->once()) + ->method('error') + ->with('Unknown error while retrieving scheduled TaskProcessing tasks', ['exception' => $exception]); + + $this->manager->expects($this->never()) + ->method('processTask'); + + $input = new ArrayInput(['--once' => true], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } + + public function testTimeoutExitsLoop(): void { + // Arrange: no providers so each iteration does nothing, but timeout=1 should exit quickly + $this->manager->method('getProviders')->willReturn([]); + + $input = new ArrayInput(['--timeout' => '1', '--interval' => '0'], $this->command->getDefinition()); + $output = new NullOutput(); + + $start = time(); + $result = $this->command->run($input, $output); + $elapsed = time() - $start; + + $this->assertSame(0, $result); + // Should have exited within a few seconds + $this->assertLessThanOrEqual(5, $elapsed); + } + + public function testProcessesFirstMatchingProvider(): void { + $taskTypeId1 = 'type_a'; + $taskTypeId2 = 'type_b'; + + $provider1 = $this->createProvider('provider_a', $taskTypeId1); + $provider2 = $this->createProvider('provider_b', $taskTypeId2); + $task = $this->createTask(7); + + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([$provider1, $provider2]); + + $this->manager->expects($this->once()) + ->method('getPreferredProvider') + ->with($taskTypeId1) + ->willReturn($provider1); + + $this->manager->expects($this->once()) + ->method('getNextScheduledTask') + ->with([$taskTypeId1]) + ->willReturn($task); + + $this->manager->expects($this->once()) + ->method('processTask') + ->with($task, $provider1) + ->willReturn(true); + + $input = new ArrayInput(['--once' => true], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } + + public function testTaskTypesWhitelistFiltersProviders(): void { + $taskTypeId1 = 'type_a'; + $taskTypeId2 = 'type_b'; + + $provider1 = $this->createProvider('provider_a', $taskTypeId1); + $provider2 = $this->createProvider('provider_b', $taskTypeId2); + $task = $this->createTask(99); + + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([$provider1, $provider2]); + + // Only type_b is whitelisted, so provider_a (type_a) must be skipped entirely + $this->manager->expects($this->once()) + ->method('getPreferredProvider') + ->with($taskTypeId2) + ->willReturn($provider2); + + $this->manager->expects($this->once()) + ->method('getNextScheduledTask') + ->with([$taskTypeId2]) + ->willReturn($task); + + $this->manager->expects($this->once()) + ->method('processTask') + ->with($task, $provider2) + ->willReturn(true); + + $input = new ArrayInput(['--once' => true, '--taskTypes' => [$taskTypeId2]], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } + + public function testTaskTypesWhitelistWithNoMatchingProviders(): void { + $provider = $this->createProvider('provider_a', 'type_a'); + + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([$provider]); + + // Whitelist does not include type_a so nothing should be processed + $this->manager->expects($this->never()) + ->method('getPreferredProvider'); + + $this->manager->expects($this->never()) + ->method('getNextScheduledTask'); + + $input = new ArrayInput(['--once' => true, '--taskTypes' => ['type_b']], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } + + public function testEmptyTaskTypesAllowsAllProviders(): void { + $taskTypeId = 'type_a'; + $provider = $this->createProvider('provider_a', $taskTypeId); + $task = $this->createTask(5); + + $this->manager->expects($this->once()) + ->method('getProviders') + ->willReturn([$provider]); + + $this->manager->expects($this->once()) + ->method('getPreferredProvider') + ->with($taskTypeId) + ->willReturn($provider); + + $this->manager->expects($this->once()) + ->method('getNextScheduledTask') + ->with([$taskTypeId]) + ->willReturn($task); + + $this->manager->expects($this->once()) + ->method('processTask') + ->with($task, $provider) + ->willReturn(true); + + // No --taskTypes option provided + $input = new ArrayInput(['--once' => true], $this->command->getDefinition()); + $output = new NullOutput(); + + $result = $this->command->run($input, $output); + + $this->assertSame(0, $result); + } +}