From c204185ae86a98994f80010abf35a190c9406739 Mon Sep 17 00:00:00 2001
From: Jake Vanderwerf <get@jakevanderwerf.ca>
Date: Sun, 12 Jul 2026 18:08:19 +0000
Subject: [PATCH] =Refactor of Integrations.php. Separated different functionality into traits that classes can use to add that functionality. Hopefully will make maintaining it a little easier. Still have to finish up, as well as refactoring the individual classes to utilize the new system.
---
inc/managers/queue/Processor.php | 133 ++++++++++++++++---------------------------
1 files changed, 50 insertions(+), 83 deletions(-)
diff --git a/inc/managers/queue/Processor.php b/inc/managers/queue/Processor.php
index a3eac9d..4178df4 100644
--- a/inc/managers/queue/Processor.php
+++ b/inc/managers/queue/Processor.php
@@ -19,54 +19,43 @@
return;
}
- $ops = $this->storage->fetchRunnable(3);
+ $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;
+ }
+ }
+ });
- foreach ($ops as $op) {
- 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
{
$progress = new Progress($op);
-
$executor = $this->registry->getExecutor($op->type) ?? $this->defaultExecutor;
$op->startedAt = current_time('mysql');
$op->state = 'processing';
-
$this->storage->saveProgress($op);
- //Check to see if we can merge first
- $mergeable = $this->registry->getMergeable($op->type);
-
- if ($mergeable) {
- $existing = $this->storage->findMergeable(
- $op->type,
- $op->userId
- );
-
- if ($existing && $mergeable->canMerge($existing, $op)) {
- $this->applyMerge($mergeable, $existing, $op);
- return;
- }
- }
-
try {
- // Check if this operation should be chunked
$chunkKey = $op->metadata['chunk_key'] ?? null;
+ // No transaction wrapping — executor handles its own
$result = $chunkKey
? $this->executeChunked($op, $executor, $progress, $chunkKey)
- : $this->storage->withTransaction(fn() => $executor->execute($op, $progress));
+ : $executor->execute($op, $progress);
$op->state = 'completed';
$op->outcome = $result->outcome;
@@ -78,7 +67,16 @@
$this->handleFailure($op, $e);
}
- $this->storage->saveFinal($op);
+ $this->saveOperation($op);
+ }
+ private function saveOperation(Operation $op): void
+ {
+ if ($op->state === 'completed') {
+ $this->storage->saveFinal($op);
+ } else {
+ // Retryable failure — save as scheduled/failed without requiring 'completed' state
+ $this->storage->save($op);
+ }
}
private function executeChunked(Operation $op, Executor $executor, Progress $progress, string|array $chunkKey): Result
@@ -116,23 +114,20 @@
}
try {
- $chunkResult = $this->storage->withTransaction(function () use ($op, $executor, $progress, $chunk, $keys, $index) {
- // Clone operation with only this chunk's data
- $chunkOp = clone $op;
- $chunkOp->requestData = array_merge(
- array_diff_key($op->requestData, array_flip($keys)),
- $chunk['data']
- );
+ $chunkOp = clone $op;
+ $chunkOp->requestData = array_merge(
+ array_diff_key($op->requestData, array_flip($keys)),
+ $chunk['data']
+ );
- // Execute this chunk
+ $executeChunk = function () use ($op, $executor, $progress, $chunkOp, $index) {
$result = $executor->execute($chunkOp, $progress);
-
- // Update progress
$op->metadata['chunk_offset'] = $index + 1;
$this->storage->saveProgress($op);
-
return $result;
- });
+ };
+
+ $chunkResult = $executeChunk();
// Merge results
if (!empty($chunkResult->result)) {
@@ -223,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
@@ -243,33 +238,6 @@
return date('Y-m-d H:i:s', time() + $delay + $jitter);
}
- private function applyMerge(
- Mergeable $mergeable,
- Operation $target,
- Operation $incoming
- ): void {
- // Safety: only merge into actively processing ops
- if ($target->state !== 'processing') {
- return;
- }
-
- $this->storage->withTransaction(function () use ($mergeable, $target, $incoming) {
- $mergeable->merge($target, $incoming);
-
- $target->dependencies[] = $incoming->id;
- $target->dependencies = array_values(array_unique($target->dependencies));
-
- $this->storage->saveProgress($target);
-
- $incoming->state = 'completed';
- $incoming->outcome = 'success';
- $incoming->completedAt = current_time('mysql');
- $incoming->result = ['merged_into' => $target->id];
-
- $this->storage->saveFinal($incoming);
- });
- }
-
private function checkResourceLimits(): bool
{
// Check memory (leave 20% buffer)
@@ -311,7 +279,7 @@
};
}
- private function hasAdequateResources(): bool
+ public function hasAdequateResources(): bool
{
// Stricter thresholds for starting (50% memory, 60s minimum time)
$memoryLimit = $this->getMemoryLimitBytes();
@@ -336,7 +304,6 @@
if (empty($op->dependencies)) {
return true;
}
-
foreach ($op->dependencies as $depId) {
$dep = $this->storage->find($depId);
--
Gitblit v1.10.0