From de317675a8069b747cb253ba3e2b5dc394ca36ef Mon Sep 17 00:00:00 2001
From: Jake Vanderwerf <get@jakevanderwerf.ca>
Date: Sat, 18 Jul 2026 21:44:51 +0000
Subject: [PATCH] =Base Integrations mainly done - doing a test with setting up the Square integration to follow
---
inc/managers/queue/Processor.php | 67 ++++++++++++++-------------------
1 files changed, 29 insertions(+), 38 deletions(-)
diff --git a/inc/managers/queue/Processor.php b/inc/managers/queue/Processor.php
index ccc4ddb..4178df4 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 {
@@ -227,17 +218,17 @@
$op->errorMessage = $e->getMessage();
- JVB()->error()->log(
- '[Queue]:processOne',
- $e->getMessage(),
- [
- 'operation_id' => $op->id,
- 'type' => $op->type,
- 'user_id' => $op->userId,
- 'retries' => $op->retries,
- ],
- $op->outcome === 'failed_permanent' ? 'critical' : 'warning'
- );
+// JVB()->error()->log(
+// '[Queue]:processOne',
+// $e->getMessage(),
+// [
+// 'operation_id' => $op->id,
+// 'type' => $op->type,
+// 'user_id' => $op->userId,
+// 'retries' => $op->retries,
+// ],
+// $op->outcome === 'failed_permanent' ? 'critical' : 'warning'
+// );
}
private function calculateBackoff(int $attempt): string
@@ -288,7 +279,7 @@
};
}
- private function hasAdequateResources(): bool
+ public function hasAdequateResources(): bool
{
// Stricter thresholds for starting (50% memory, 60s minimum time)
$memoryLimit = $this->getMemoryLimitBytes();
--
Gitblit v1.10.0