fix(taskprocessing): claim tasks atomically in worker via lockTask()

Signed-off-by: bygadd <bygadd@gmail.com>

Author: bygadd

core/Command/TaskProcessing/WorkerCommand.php

@@ -165,17 +165,31 @@ private function processNextTask(OutputInterface $output, array $taskTypes = [])
 		// 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;
-		}
+		$taskIdsToIgnore = [];
+		while (true) {
+			try {
+				$task = $this->taskProcessingManager->getNextScheduledTask(array_keys($eligibleProviders), $taskIdsToIgnore);
+			} 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];
+			$taskTypeId = $task->getTaskTypeId();
+			if (!isset($eligibleProviders[$taskTypeId])) {
+				$taskIdsToIgnore[] = (int)$task->getId();
+				continue;
+			}
+			$provider = $eligibleProviders[$taskTypeId];
+
+			// Atomically claim the task; if another worker grabbed it between
+			// the SELECT above and now, lockTask() returns false -> skip it.
+			if ($this->taskProcessingManager->lockTask($task)) {
+				break;
+			}
+			$taskIdsToIgnore[] = (int)$task->getId();
+		}
 
 		$output->writeln(
 			'Processing task ' . $task->getId() . ' of type ' . $taskTypeId . ' with provider ' . $provider->getId(),