From 97e7c319d656a5f05489ca996e249e7359303d4d Mon Sep 17 00:00:00 2001
From: Jake Vanderwerf <get@jakevanderwerf.ca>
Date: Sun, 31 May 2026 22:42:33 +0000
Subject: [PATCH] =Jakevan edits done?
---
inc/managers/queue/Processor.php | 43 +++++++++++++++++--------------------------
1 files changed, 17 insertions(+), 26 deletions(-)
diff --git a/inc/managers/queue/Processor.php b/inc/managers/queue/Processor.php
index ccc4ddb..f54fadd 100644
--- a/inc/managers/queue/Processor.php
+++ b/inc/managers/queue/Processor.php
@@ -14,48 +14,39 @@
public function run(): void
{
- if (get_transient(BASE.'queue_running')) {
- return;
- }
- set_transient(BASE.'queue_running', true, 60);
if (!$this->hasAdequateResources()) {
error_log('[Processor] Insufficient resources to start processing');
return;
}
- $ops = $this->storage->fetchRunnable();
- if (empty($ops)) {
- return;
- }
- foreach ($ops as $op) {
- if ($op->state === 'completed') {
- return;
+ $op = null;
+ $this->storage->withTransaction(function() use (&$op) {
+ $candidates = $this->storage->fetchRunnable();
+ foreach ($candidates as $candidate) {
+ if ($candidate->state === 'completed') continue;
+ if (!$this->dependenciesSatisfied($candidate)) continue;
+ if ($this->storage->markProcessing($candidate->id)) {
+ $op = $candidate;
+ break;
+ }
}
- if (!$this->dependenciesSatisfied($op)) {
- continue;
- }
- if (!$this->storage->markProcessing($op->id)) {
- continue;
- }
- $this->processOne($op);
- usleep(10000);
- }
+ });
+
+ if (!$op) return;
+
+ $this->processOne($op);
+ usleep(10000);
$this->storage->invalidateQueueCache();
}
+
private function processOne(Operation $op): void
{
- if (get_transient(BASE.$op->id)) {
- return;
- }
- set_transient(BASE.$op->id, true, 500);
$progress = new Progress($op);
-
$executor = $this->registry->getExecutor($op->type) ?? $this->defaultExecutor;
$op->startedAt = current_time('mysql');
$op->state = 'processing';
-
$this->storage->saveProgress($op);
try {
--
Gitblit v1.10.0