From c45b0798568d2aa6b388ecbd9bffa5de7e9a34ae Mon Sep 17 00:00:00 2001
From: Jake Vanderwerf <get@jakevanderwerf.ca>
Date: Tue, 10 Feb 2026 02:20:38 +0000
Subject: [PATCH] =minor fixes to schema system
---
inc/managers/queue/Processor.php | 293 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++--
1 files changed, 281 insertions(+), 12 deletions(-)
diff --git a/inc/managers/queue/Processor.php b/inc/managers/queue/Processor.php
index 0f02650..a3eac9d 100644
--- a/inc/managers/queue/Processor.php
+++ b/inc/managers/queue/Processor.php
@@ -10,33 +10,63 @@
private Storage $storage,
private Executor $defaultExecutor,
private TypeRegistry $registry,
- private Locker $locker
) {}
public function run(): void
{
- $this->locker->withLock(function () {
- $ops = $this->storage->fetchRunnable(10);
+ if (!$this->hasAdequateResources()) {
+ error_log('[Processor] Insufficient resources to start processing');
+ return;
+ }
- foreach ($ops as $op) {
- if (!$this->storage->markProcessing($op->id)) {
- continue;
- }
+ $ops = $this->storage->fetchRunnable(3);
- $this->processOne($op);
+ foreach ($ops as $op) {
+ if (!$this->dependenciesSatisfied($op)) {
+ continue;
}
+ if (!$this->storage->markProcessing($op->id)) {
+ continue;
+ }
+ $this->processOne($op);
+ usleep(10000);
+ }
- $this->storage->invalidateQueueCache();
- });
+ $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 {
- $result = $executor->execute($op, $progress);
+ // Check if this operation should be chunked
+ $chunkKey = $op->metadata['chunk_key'] ?? null;
+
+ $result = $chunkKey
+ ? $this->executeChunked($op, $executor, $progress, $chunkKey)
+ : $this->storage->withTransaction(fn() => $executor->execute($op, $progress));
$op->state = 'completed';
$op->outcome = $result->outcome;
@@ -44,10 +74,134 @@
$op->completedAt = current_time('mysql');
} catch (\Throwable $e) {
+ error_log("[Processor] Exception caught: " . $e->getMessage());
$this->handleFailure($op, $e);
}
- $this->storage->save($op);
+ $this->storage->saveFinal($op);
+ }
+
+ private function executeChunked(Operation $op, Executor $executor, Progress $progress, string|array $chunkKey): Result
+ {
+ $keys = (array) $chunkKey;
+ $chunkSize = $op->metadata['chunk_size'] ?? 10;
+ $offset = $op->metadata['chunk_offset'] ?? 0;
+ $chunks = $this->buildChunks($op->requestData, $keys, $chunkSize);
+ $totalChunks = count($chunks);
+
+ $allResults = [];
+
+ foreach ($chunks as $index => $chunk) {
+ if ($index < $offset) {
+ continue;
+ }
+
+ // Resource check before each chunk
+ if (!$this->checkResourceLimits()) {
+ error_log("[Processor] Resource limits reached, pausing at chunk {$index}");
+
+ $op->metadata['chunk_offset'] = $index;
+ $op->state = 'scheduled';
+ $op->scheduledAt = date('Y-m-d H:i:s', time() + 5);
+ $this->storage->save($op);
+
+ return new Result(
+ outcome: 'pending',
+ result: [
+ 'paused_at_chunk' => $index,
+ 'total_chunks' => $totalChunks,
+ 'partial_results' => $allResults,
+ ]
+ );
+ }
+
+ 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']
+ );
+
+ // Execute this chunk
+ $result = $executor->execute($chunkOp, $progress);
+
+ // Update progress
+ $op->metadata['chunk_offset'] = $index + 1;
+ $this->storage->saveProgress($op);
+
+ return $result;
+ });
+
+ // Merge results
+ if (!empty($chunkResult->result)) {
+ if (is_array($chunkResult->result)) {
+ $allResults = array_merge_recursive($allResults, $chunkResult->result);
+ } else {
+ $allResults = $chunkResult->result;
+ }
+ }
+
+ } catch (\Throwable $e) {
+ error_log("[Processor] Chunk {$index} failed: " . $e->getMessage());
+
+ // Record failed items from this chunk
+ foreach ($chunk['data'] as $key => $items) {
+ foreach ($items as $item) {
+ $progress->failItem($item, $e->getMessage());
+ }
+ }
+
+ // Continue to next chunk rather than failing entire operation
+ // Remove this if you want fail-fast behavior
+ }
+
+ // Delay between chunks (skip after last chunk)
+ if ($index < $totalChunks - 1) {
+ usleep(50000); // 50ms
+ }
+ }
+
+ // Determine final outcome
+ $outcome = 'success';
+ if (!empty($op->failedItems)) {
+ $failedCount = count($op->failedItems);
+ $outcome = $failedCount === $op->totalItems ? 'failed' : 'partial';
+ }
+
+ return new Result(
+ outcome: $outcome,
+ result: $allResults
+ );
+ }
+
+ private function buildChunks(array $data, array $keys, int $chunkSize): array
+ {
+ $chunks = [];
+
+ foreach ($keys as $key) {
+ if (!isset($data[$key]) || !is_array($data[$key])) {
+ continue;
+ }
+
+ $items = $data[$key];
+ $itemChunks = array_chunk($items, $chunkSize, true); // preserve keys
+
+ foreach ($itemChunks as $index => $chunkItems) {
+ if (!isset($chunks[$index])) {
+ $chunks[$index] = [
+ 'data' => [],
+ 'count' => 0
+ ];
+ }
+
+ $chunks[$index]['data'][$key] = $chunkItems;
+ $chunks[$index]['count'] += count($chunkItems);
+ }
+ }
+
+ return $chunks;
}
private function handleFailure(Operation $op, \Throwable $e): void
@@ -88,4 +242,119 @@
$jitter = rand(0, (int)($delay * 0.1));
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)
+ $memoryLimit = $this->getMemoryLimitBytes();
+ $memoryUsage = memory_get_usage(true);
+ if ($memoryUsage > $memoryLimit * 0.8) {
+ error_log('[Processor] Memory limit approaching, pausing');
+ return false;
+ }
+
+ // Check time (leave 30s buffer for cleanup)
+ $maxTime = (int) ini_get('max_execution_time');
+ if ($maxTime > 0) {
+ $elapsed = microtime(true) - $_SERVER['REQUEST_TIME_FLOAT'];
+ if ($elapsed > $maxTime - 30) {
+ error_log('[Processor] Time limit approaching, pausing');
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private function getMemoryLimitBytes(): int
+ {
+ $limit = ini_get('memory_limit');
+ if ($limit === '-1') {
+ return PHP_INT_MAX;
+ }
+
+ $unit = strtolower(substr($limit, -1));
+ $value = (int) $limit;
+
+ return match($unit) {
+ 'g' => $value * 1024 * 1024 * 1024,
+ 'm' => $value * 1024 * 1024,
+ 'k' => $value * 1024,
+ default => $value,
+ };
+ }
+
+ private function hasAdequateResources(): bool
+ {
+ // Stricter thresholds for starting (50% memory, 60s minimum time)
+ $memoryLimit = $this->getMemoryLimitBytes();
+ $memoryUsage = memory_get_usage(true);
+ if ($memoryUsage > $memoryLimit * 0.5) {
+ return false;
+ }
+
+ $maxTime = (int) ini_get('max_execution_time');
+ if ($maxTime > 0) {
+ $elapsed = microtime(true) - $_SERVER['REQUEST_TIME_FLOAT'];
+ if ($maxTime - $elapsed < 60) { // Need at least 60s
+ return false;
+ }
+ }
+
+ 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