Merge branch 'master' into main

Author: dereuromark

config/Migrations/20240307154751_MigrationQueueInitV8.php

@@ -9,9 +9,6 @@ class MigrationQueueInitV8 extends BaseMigration {
 	/**
 	 * Up Method.
 	 *
-	 * More information on this method is available here:
-	 * https://book.cakephp.org/phinx/0/en/migrations.html#the-up-method
-	 *
 	 * @return void
 	 */
 	public function up(): void {

config/Migrations/20250508121432_MigrationQueueMemory.php

@@ -8,9 +8,6 @@ class MigrationQueueMemory extends BaseMigration {
 	/**
 	 * Change Method.
 	 *
-	 * More information on this method is available here:
-	 * https://book.cakephp.org/migrations/4/en/migrations.html#the-change-method
-	 *
 	 * @return void
 	 */
 	public function change(): void {

config/Migrations/20251119005411_MigrationQueueIndexOptimization.php

@@ -0,0 +1,51 @@
+<?php
+declare(strict_types=1);
+
+use Migrations\BaseMigration;
+
+class MigrationQueueIndexOptimization extends BaseMigration {
+
+	/**
+	 * Change Method.
+	 *
+	 * @return void
+	 */
+	public function change(): void {
+		$this->table('queued_jobs')
+			// Main queue fetch optimization - CRITICAL
+			// Covers the primary requestJob() query which filters by completed IS NULL
+			// and orders by priority, age (calculated from notbefore), and id
+			->addIndex(
+				['completed', 'priority', 'notbefore', 'id'],
+				[
+					'name' => 'queue_fetch_optimization',
+				],
+			)
+			// Cleanup queries optimization
+			// Used by flushFailedJobs() which filters by fetched < threshold
+			->addIndex(
+				['fetched'],
+				[
+					'name' => 'fetched',
+				],
+			)
+			// Stats and job type queries optimization
+			// Used by getStats() and various job_task lookups
+			->addIndex(
+				['job_task', 'completed'],
+				[
+					'name' => 'job_task_completed',
+				],
+			)
+			// Worker tracking optimization
+			// Used in requestJob() for cost and unique constraints
+			->addIndex(
+				['workerkey'],
+				[
+					'name' => 'workerkey',
+				],
+			)
+			->update();
+	}
+
+}

phpstan.neon

@@ -14,7 +14,6 @@ parameters:
 		- identifier: missingType.iterableValue
 		- identifier: missingType.generics
 		- identifier: trait.unused
-		- '#Negated boolean expression is always false.#'
 		- '#Parameter \#1 \$.+ of function call_user_func_array expects .+, array.+ given.#'
 
 includes:

src/Controller/Admin/QueuedJobsController.php

@@ -142,13 +142,27 @@ public function import() {
 			/** @var \Laminas\Diactoros\UploadedFile|null $file */
 			$file = $this->request->getData('file');
 			if ($file && $file->getError() == UPLOAD_ERR_OK && $file->getSize() > 0) {
+				$clientMediaType = $file->getClientMediaType();
+				if ($clientMediaType !== 'application/json') {
+					throw new RuntimeException('Only JSON files are allowed');
+				}
+
 				$content = file_get_contents($file->getStream()->getMetadata('uri'));
 				if ($content === false) {
 					throw new RuntimeException('Cannot parse file');
 				}
+
 				$json = json_decode($content, true);
+				if (json_last_error() !== JSON_ERROR_NONE) {
+					throw new RuntimeException('Invalid JSON: ' . json_last_error_msg());
+				}
+
 				if (!$json || empty($json['queuedJob'])) {
-					throw new RuntimeException('Invalid JSON content');
+					throw new RuntimeException('Invalid JSON content: missing queuedJob data');
+				}
+
+				if (!is_array($json['queuedJob'])) {
+					throw new RuntimeException('Invalid JSON structure: queuedJob must be an array');
 				}
 
 				$data = $json['queuedJob'];

src/Model/Table/QueuedJobsTable.php

@@ -613,7 +613,7 @@ public function requestJob(array $tasks, array $groups = [], array $types = []):
 
 						break;
 					case static::DRIVER_SQLITE:
-						//TODO
+						$tmp['strftime("%s", "now") >='] = $this->rateHistory[$tmp['job_task']] + $task['rate'];
 
 						break;
 				}
@@ -632,8 +632,10 @@ public function requestJob(array $tasks, array $groups = [], array $types = []):
 
 					break;
 				case static::DRIVER_SQLSERVER:
-					// the ORM does not support ROW locking at the moment
-					// TODO
+					// Row-level locking (ROWLOCK, UPDLOCK hints) not supported by CakePHP ORM.
+					// SQLServer uses default isolation level (READ COMMITTED) with automatic
+					// row locking on UPDATE. This may allow race conditions in high-concurrency
+					// scenarios. Consider using application-level locking or advisory locks if needed.
 
 					break;
 				case static::DRIVER_SQLITE:
@@ -925,8 +927,10 @@ public function cleanOldJobs(): int {
 			return 0;
 		}
 
+		$threshold = (new DateTime())->subSeconds((int)Configure::read('Queue.cleanuptimeout'));
+
 		return $this->deleteAll([
-			'completed <' => time() - (int)Configure::read('Queue.cleanuptimeout'),
+			'completed <' => $threshold,
 		]);
 	}
 
@@ -979,7 +983,7 @@ public function key(): string {
 		if ($this->_key !== null) {
 			return $this->_key;
 		}
-		$this->_key = sha1(microtime() . mt_rand(100, 999));
+		$this->_key = bin2hex(random_bytes(20));
 		if (!$this->_key) {
 			throw new RuntimeException('Invalid key generated');
 		}

src/Queue/Task/ExecuteTask.php

@@ -10,6 +10,7 @@
 
 use Cake\Console\CommandInterface;
 use Cake\Console\ConsoleIo;
+use Cake\Core\Configure;
 use Cake\Log\LogTrait;
 use Queue\Model\QueueException;
 use Queue\Queue\AddInterface;
@@ -85,6 +86,10 @@ public function run(array $data, int $jobId): void {
 			'accepted' => [CommandInterface::CODE_SUCCESS],
 		];
 
+		if (!$data['escape'] && !Configure::read('debug')) {
+			throw new QueueException('Command escaping must be enabled when debug mode is off for security reasons');
+		}
+
 		$command = $data['command'];
 		if ($data['escape']) {
 			$command = escapeshellcmd($data['command']);