Merge pull request #59015 from nextcloud/copilot/add-taskprocessing-worker-command

feat(taskprocessing): add worker command for synchronous task processing

core/Command/TaskProcessing/WorkerCommand.php

@@ -0,0 +1,182 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+namespace OC\Core\Command\TaskProcessing;
+
+use OC\Core\Command\Base;
+use OC\Core\Command\InterruptedException;
+use OCP\TaskProcessing\Exception\Exception;
+use OCP\TaskProcessing\Exception\NotFoundException;
+use OCP\TaskProcessing\IManager;
+use OCP\TaskProcessing\ISynchronousProvider;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class WorkerCommand extends Base {
+	public function __construct(
+		private readonly IManager $taskProcessingManager,
+		private readonly LoggerInterface $logger,
+	) {
+		parent::__construct();
+	}
+
+	protected function configure(): void {
+		$this
+			->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). You should regularly (e.g. every 5 minutes) restart this worker by using this option to make sure it picks up configuration changes.',
+				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<string> $taskTypes */
+		$taskTypes = $input->getOption('taskTypes');
+
+		if ($timeout > 0) {
+			$output->writeln('<info>Task processing worker will stop after ' . $timeout . ' seconds</info>');
+		}
+
+		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('<info>Task processing worker stopped</info>');
+				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<string> $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<string, ISynchronousProvider> $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;
+	}
+}

core/register_command.php

@@ -91,6 +91,7 @@ use OC\Core\Command\SystemTag\Edit;
 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 @@ if ($config->getSystemValueBool('installed', false)) {
 	$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));

lib/composer/composer/autoload_classmap.php

@@ -1423,6 +1423,7 @@ return array(
     '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',

lib/composer/composer/autoload_static.php

@@ -1464,6 +1464,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',

tests/Core/Command/TaskProcessing/WorkerCommandTest.php

@@ -0,0 +1,360 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+namespace Tests\Core\Command\TaskProcessing;
+
+use OC\Core\Command\TaskProcessing\WorkerCommand;
+use OCP\TaskProcessing\Exception\Exception;
+use OCP\TaskProcessing\Exception\NotFoundException;
+use OCP\TaskProcessing\IManager;
+use OCP\TaskProcessing\ISynchronousProvider;
+use OCP\TaskProcessing\Task;
+use PHPUnit\Framework\MockObject\MockObject;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Console\Input\ArrayInput;
+use Symfony\Component\Console\Output\NullOutput;
+use Test\TestCase;
+
+class WorkerCommandTest extends TestCase {
+	private IManager&MockObject $manager;
+	private LoggerInterface&MockObject $logger;
+	private WorkerCommand $command;
+
+	protected function setUp(): void {
+		parent::setUp();
+
+		$this->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 and optional task type.
+	 */
+	private function createTask(int $id, string $type = 'test_task_type'): Task {
+		$task = new 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 testProcessesCorrectProviderForReturnedTaskType(): void {
+		$taskTypeId1 = 'type_a';
+		$taskTypeId2 = 'type_b';
+
+		$provider1 = $this->createProvider('provider_a', $taskTypeId1);
+		$provider2 = $this->createProvider('provider_b', $taskTypeId2);
+		// Task has type_a, so provider1 must be chosen to process it
+		$task = $this->createTask(7, $taskTypeId1);
+
+		$this->manager->expects($this->once())
+			->method('getProviders')
+			->willReturn([$provider1, $provider2]);
+
+		// Both providers are eligible, so getPreferredProvider is called for each
+		$this->manager->expects($this->exactly(2))
+			->method('getPreferredProvider')
+			->willReturnMap([
+				[$taskTypeId1, $provider1],
+				[$taskTypeId2, $provider2],
+			]);
+
+		// All eligible task types are passed in a single query
+		$this->manager->expects($this->once())
+			->method('getNextScheduledTask')
+			->with($this->equalTo([$taskTypeId1, $taskTypeId2]))
+			->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, $taskTypeId2);
+
+		$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, $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])
+			->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);
+	}
+}