From 27fb820ae9081fb56957cf75e79eccd8a99edd52 Mon Sep 17 00:00:00 2001
From: Jake Vanderwerf <get@jakevanderwerf.ca>
Date: Sat, 14 Feb 2026 19:14:48 +0000
Subject: [PATCH] =minor css changes, queue merge logic restructured, and double checking changes actually changed in cRUD.js before sending to server
---
inc/managers/queue/Processor.php | 118 ++++++++++++++++++++++++++++------------------------------
1 files changed, 57 insertions(+), 61 deletions(-)
diff --git a/inc/managers/queue/Processor.php b/inc/managers/queue/Processor.php
index d743f77..e3d7011 100644
--- a/inc/managers/queue/Processor.php
+++ b/inc/managers/queue/Processor.php
@@ -19,14 +19,20 @@
return;
}
- $ops = $this->storage->fetchRunnable(3);
-
- $lastOpId = null;
+ $ops = $this->storage->fetchRunnable();
+ if (empty($ops)) {
+ return;
+ }
foreach ($ops as $op) {
+ if ($op->state === 'completed') {
+ return;
+ }
+ if (!$this->dependenciesSatisfied($op)) {
+ continue;
+ }
if (!$this->storage->markProcessing($op->id)) {
continue;
}
- $lastOpId = $op->id;
$this->processOne($op);
usleep(10000);
}
@@ -44,28 +50,13 @@
$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;
@@ -77,7 +68,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
@@ -115,23 +115,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)) {
@@ -242,33 +239,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)
@@ -329,4 +299,30 @@
return true;
}
+
+ private function dependenciesSatisfied(Operation $op): bool
+ {
+ if (empty($op->dependencies)) {
+ return true;
+ }
+ foreach ($op->dependencies as $depId) {
+ $dep = $this->storage->find($depId);
+
+ // Missing dependency = block (or decide to ignore; your call)
+ if (!$dep) {
+ return false;
+ }
+
+ if ($dep->state !== 'completed') {
+ return false;
+ }
+
+ if (!in_array($dep->outcome, ['success', 'partial'], true)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
}
--
Gitblit v1.10.0