fix(taskprocessing): guard lockTask on scheduled, record started_at, Oracle fallback

Address review feedback (@marcelklehr, Copilot):
- lockTask claims only SCHEDULED tasks (was status != RUNNING) and stamps
  started_at in the same atomic UPDATE, so a finished task cannot be re-claimed
  and the external-provider claim path records started_at as well.
- claimWithBoundedRetry re-reads after lockTask instead of a follow-up UPDATE.
- Oracle joins SQLite on the bounded-retry fallback: Oracle cannot combine a
  row-limiting clause with FOR UPDATE (ORA-02014), which failed the claim tests
  on Oracle CI.
- Reword the worker docblock/comments to "prefer oldest available" (parallel
  SKIP LOCKED does not guarantee a strict global order).
- Add a regression test that lockTask does not resurrect a finished task.

Signed-off-by: Yoan Bozhilov <bygadd@gmail.com>
Assisted-by: Claude Code:claude-opus-4-8

Author: bygadd

core/Command/TaskProcessing/WorkerCommand.php

@@ -118,9 +118,10 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 	 *
 	 * To avoid starvation, all eligible task types are first collected and then
 	 * the oldest scheduled task across all of them is claimed in a single atomic
-	 * query (FOR UPDATE SKIP LOCKED, with a SQLite fallback). This ensures tasks
-	 * are processed in the order they were scheduled, regardless of which provider
-	 * handles them, and guarantees no two workers ever claim the same task.
+	 * query (FOR UPDATE SKIP LOCKED, with a SQLite fallback). Each claim prefers the
+	 * oldest available scheduled task -- under parallel workers SKIP LOCKED skips rows
+	 * another worker has locked, so this reduces starvation rather than guaranteeing a
+	 * strict global processing order -- and no two workers ever claim the same task.
 	 *
 	 * @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
@@ -165,9 +166,10 @@ private function processNextTask(OutputInterface $output, array $taskTypes = [])
 		// Atomically claim the oldest scheduled task across all eligible task types in
 		// one query. SELECT ... FOR UPDATE SKIP LOCKED (with a SQLite fallback) both
 		// fetches and marks the task RUNNING, so multiple workers never claim the same
-		// task and no per-worker ignore-list / retry loop is needed. This also naturally
-		// prevents starvation: regardless of how many tasks one provider has queued,
-		// another provider's older tasks are picked up first.
+		// task and no per-worker ignore-list / retry loop is needed. This also reduces
+		// starvation: each claim prefers the oldest available task, so a provider with a
+		// large queue does not indefinitely block another provider's older tasks (though a
+		// worker may claim a newer task while an older one is locked by another worker).
 		try {
 			$task = $this->taskProcessingManager->claimNextScheduledTask(array_keys($eligibleProviders));
 		} catch (Exception $e) {

lib/private/TaskProcessing/Db/TaskMapper.php

@@ -90,11 +90,15 @@ public function findOldestScheduledByType(array $taskTypes, array $taskIdsToIgno
 	 * followed by a guarded UPDATE to RUNNING. Concurrent workers skip rows already
 	 * locked by another transaction, so no two workers ever claim the same task.
 	 *
-	 * SQLite does not support SKIP LOCKED (verified: Doctrine throws "Operation
-	 * 'SKIP LOCKED' is not supported by platform"), so we feature-detect via the DB
-	 * provider and fall back to the existing bounded {@see lockTask} retry, which is
-	 * still safe because the UPDATE ... WHERE status = SCHEDULED is itself atomic and
-	 * SQLite serialises writers.
+	 * Two databases cannot use the SKIP LOCKED path and fall back to a bounded
+	 * lock-and-retry claim instead:
+	 *   - SQLite has no SKIP LOCKED (Doctrine throws "Operation 'SKIP LOCKED' is not
+	 *     supported by platform").
+	 *   - Oracle cannot combine a row-limiting clause with FOR UPDATE: the LIMIT is
+	 *     emulated with a ROWNUM sub-select, and selecting FOR UPDATE from that derived
+	 *     view raises ORA-02014.
+	 * The fallback is still safe because the UPDATE ... WHERE status = SCHEDULED is itself
+	 * atomic (SQLite additionally serialises writers).
 	 *
 	 * A task is only ever transitioned SCHEDULED -> RUNNING here; it is never marked
 	 * FAILED by claiming. If the task cannot be claimed (none scheduled, or it was
@@ -105,8 +109,10 @@ public function findOldestScheduledByType(array $taskTypes, array $taskIdsToIgno
 	 * @throws Exception
 	 */
 	public function claimOldestScheduledTask(array $taskTypes): ?Task {
-		if ($this->db->getDatabaseProvider() === IDBConnection::PLATFORM_SQLITE) {
-			// SKIP LOCKED is unsupported on SQLite: fall back to the bounded lock-and-retry claim.
+		$provider = $this->db->getDatabaseProvider();
+		// SKIP LOCKED is unusable on SQLite (unsupported) and Oracle (LIMIT + FOR UPDATE =>
+		// ORA-02014): both fall back to the bounded lock-and-retry claim.
+		if ($provider === IDBConnection::PLATFORM_SQLITE || $provider === IDBConnection::PLATFORM_ORACLE) {
 			return $this->claimWithBoundedRetry($taskTypes);
 		}
 
@@ -184,7 +190,7 @@ private function claimWithSkipLocked(array $taskTypes): ?Task {
 	}
 
 	/**
-	 * Fallback claim for databases without SKIP LOCKED (SQLite).
+	 * Fallback claim for databases that cannot use the SKIP LOCKED path (SQLite, Oracle).
 	 *
 	 * Repeatedly fetches the oldest scheduled task and attempts the atomic
 	 * UPDATE ... WHERE status = SCHEDULED. Tasks lost to another worker are added to a
@@ -207,19 +213,9 @@ private function claimWithBoundedRetry(array $taskTypes): ?Task {
 			}
 
 			if ($this->lockTask($task) !== 0) {
-				$task->setStatus(\OCP\TaskProcessing\Task::STATUS_RUNNING);
-				// Record the start time at claim time. lockTask only flips the status (and is
-				// shared with other callers), so persist started_at with a targeted follow-up
-				// UPDATE rather than changing lockTask's behaviour. The worker receives the task
-				// already RUNNING, so Manager::setTaskStatus would otherwise never write it.
-				$startedAt = $this->timeFactory->now()->getTimestamp();
-				$update = $this->db->getQueryBuilder();
-				$update->update($this->tableName)
-					->set('started_at', $update->createPositionalParameter($startedAt, IQueryBuilder::PARAM_INT))
-					->where($update->expr()->eq('id', $update->createPositionalParameter($task->getId(), IQueryBuilder::PARAM_INT)));
-				$update->executeStatement();
-				$task->setStartedAt($startedAt);
-				return $task;
+				// lockTask atomically flipped SCHEDULED -> RUNNING and stamped started_at.
+				// Re-read so the returned task reflects the persisted status and started_at.
+				return $this->find($task->getId());
 			}
 
 			// Another worker took it; skip this id and try the next oldest.
@@ -376,12 +372,25 @@ public function update(Entity $entity): Entity {
 		return parent::update($entity);
 	}
 
+	/**
+	 * Atomically claim a task by transitioning it SCHEDULED -> RUNNING.
+	 *
+	 * The UPDATE is guarded on `status = SCHEDULED` so a task another worker has already
+	 * finished (SUCCESSFUL/FAILED) between a caller's SELECT and this UPDATE can never be
+	 * re-claimed and processed twice. started_at is stamped in the same statement: the
+	 * worker receives the task already RUNNING, so the later SCHEDULED -> RUNNING edge in
+	 * Manager::setTaskStatus (which used to set started_at) no longer fires.
+	 *
+	 * @return int Number of rows updated: 1 if the task was claimed, 0 if it was no longer scheduled.
+	 */
 	public function lockTask(Entity $entity): int {
+		$startedAt = $this->timeFactory->now()->getTimestamp();
 		$qb = $this->db->getQueryBuilder();
 		$qb->update($this->tableName)
 			->set('status', $qb->createPositionalParameter(\OCP\TaskProcessing\Task::STATUS_RUNNING, IQueryBuilder::PARAM_INT))
+			->set('started_at', $qb->createPositionalParameter($startedAt, IQueryBuilder::PARAM_INT))
 			->where($qb->expr()->eq('id', $qb->createPositionalParameter($entity->getId(), IQueryBuilder::PARAM_INT)))
-			->andWhere($qb->expr()->neq('status', $qb->createPositionalParameter(2, IQueryBuilder::PARAM_INT)));
+			->andWhere($qb->expr()->eq('status', $qb->createPositionalParameter(\OCP\TaskProcessing\Task::STATUS_SCHEDULED, IQueryBuilder::PARAM_INT)));
 		try {
 			return $qb->executeStatement();
 		} catch (Exception) {

tests/lib/TaskProcessing/TaskProcessingTest.php

@@ -1757,4 +1757,29 @@ public function testClaimRecordsStartedAt(): void {
 		self::assertGreaterThanOrEqual($before, $persisted->getStartedAt());
 		self::assertLessThanOrEqual($after, $persisted->getStartedAt());
 	}
+
+	public function testLockTaskDoesNotResurrectFinishedTask(): void {
+		// Regression guard for the lockTask claim path (used by the SQLite fallback and the
+		// external-provider API claim). lockTask must only ever transition SCHEDULED -> RUNNING.
+		// If another worker finished a task (SUCCESSFUL/FAILED) between the SELECT and this
+		// UPDATE, lockTask must NOT flip it back to RUNNING -- otherwise a completed task is
+		// resurrected and processed twice. (The previous `status != RUNNING` guard let a
+		// SUCCESSFUL/FAILED row be re-locked.)
+		$this->registerTextToTextProvider();
+
+		$task = new Task(TextToText::ID, ['input' => 'Hello'], 'test', null);
+		$this->manager->scheduleTask($task);
+		$id = $task->getId();
+
+		// Simulate another worker having already finished the task.
+		$entity = $this->taskMapper->find($id);
+		$entity->setStatus(Task::STATUS_SUCCESSFUL);
+		$this->taskMapper->update($entity);
+
+		// Attempting to claim the (now SUCCESSFUL) task must be a no-op.
+		$affected = $this->taskMapper->lockTask($this->taskMapper->find($id));
+
+		self::assertSame(0, $affected, 'lockTask must not claim a task that is no longer SCHEDULED');
+		self::assertEquals(Task::STATUS_SUCCESSFUL, $this->manager->getTask($id)->getStatus());
+	}
 }