fixed runningJobs modified condition (#433)

* fixed runningJos modified condition

* fixed worker timeout

* fixed cs errors

* fixed cleanEndedProcesses

* fixed cleanEndedProcesses

* fixed cleanEndedProcesses

* fixed cleanEndedProcesses

* fixed cleanEndedProcesses

* fixed cleanEndedProcesses

---------

Co-authored-by: Gustavo <[email protected]>

Author: GustavoPeixoto

src/Model/Table/QueueProcessesTable.php

@@ -3,6 +3,7 @@
 namespace Queue\Model\Table;
 
 use Cake\Core\Configure;
+use Cake\Database\Expression\QueryExpression;
 use Cake\I18n\FrozenTime;
 use Cake\ORM\Table;
 use Cake\ORM\TableRegistry;
@@ -189,10 +190,12 @@ public function remove(string $pid): void {
 	 * @return int
 	 */
 	public function cleanEndedProcesses(): int {
-		$timeout = Config::defaultworkertimeout();
-		$thresholdTime = (new FrozenTime())->subSeconds($timeout);
+		$activeProcesses = $this->findActive()->select(['id']);
+		$ids = $activeProcesses->all()->extract('id')->toArray();
 
-		return $this->deleteAll(['modified <' => $thresholdTime]);
+		return $this->deleteAll(function (QueryExpression $exp) use ($ids): QueryExpression {
+			return $exp->notIn('id', count($ids) > 0 ? $ids : [-1]);
+		});
 	}
 
 	/**

src/Model/Table/QueuedJobsTable.php

@@ -521,7 +521,7 @@ public function requestJob(array $tasks, array $groups = [], array $types = [])
 			$constraintJobs = array_keys($costConstraints + $uniqueConstraints);
 			$runningJobs = $this->find('queued')
 				->contain(['WorkerProcesses'])
-				->where(['QueuedJobs.job_task IN' => $constraintJobs, 'QueuedJobs.workerkey IS NOT' => null, 'QueuedJobs.workerkey !=' => $this->_key, 'WorkerProcesses.modified >' => Config::defaultworkertimeout()])
+				->where(['QueuedJobs.job_task IN' => $constraintJobs, 'QueuedJobs.workerkey IS NOT' => null, 'QueuedJobs.workerkey !=' => $this->_key, 'WorkerProcesses.modified >' => (new FrozenTime())->subSeconds(Config::defaultworkertimeout())])
 				->all()
 				->toArray();
 		}

src/Queue/Config.php

@@ -15,7 +15,12 @@ class Config {
 	 * @return int
 	 */
 	public static function defaultworkertimeout() {
-		return Configure::read('Queue.defaultworkertimeout', 600); // 10min
+		$timeout = Configure::read('Queue.defaultworkertimeout', 600); // 10min
+		if ($timeout <= 0) {
+			throw new InvalidArgumentException('Queue.defaultworkertimeout is less or eqaul than zero. Indefinite running of workers is not supported.');
+		}
+
+		return $timeout;
 	}
 
 	/**