mirror of
https://github.com/getgrav/grav.git
synced 2026-07-06 14:39:25 +02:00
Andy Miller
@@ -1,545 +0,0 @@
|
||||
<?php
|
||||
|
||||
/**
|
||||
* @package Grav\Common\Scheduler
|
||||
*
|
||||
* @copyright Copyright (c) 2015 - 2025 Trilby Media, LLC. All rights reserved.
|
||||
* @license MIT License; see LICENSE file for details.
|
||||
*/
|
||||
|
||||
namespace Grav\Common\Scheduler;
|
||||
|
||||
use DateTime;
|
||||
use Exception;
|
||||
use RuntimeException;
|
||||
use Symfony\Component\Process\Process;
|
||||
|
||||
/**
|
||||
* Enhanced Job class with modern features
|
||||
*
|
||||
* @package Grav\Common\Scheduler
|
||||
*/
|
||||
class ModernJob extends Job
|
||||
{
|
||||
/** @var int */
|
||||
protected $maxAttempts = 3;
|
||||
|
||||
/** @var int */
|
||||
protected $retryCount = 0;
|
||||
|
||||
/** @var int */
|
||||
protected $retryDelay = 60; // seconds
|
||||
|
||||
/** @var string */
|
||||
protected $retryStrategy = 'exponential'; // 'linear' or 'exponential'
|
||||
|
||||
/** @var float */
|
||||
protected $executionStartTime;
|
||||
|
||||
/** @var float */
|
||||
protected $executionTime = 0;
|
||||
|
||||
/** @var int */
|
||||
protected $timeout = 300; // 5 minutes default
|
||||
|
||||
/** @var array */
|
||||
protected $dependencies = [];
|
||||
|
||||
/** @var array */
|
||||
protected $chainedJobs = [];
|
||||
|
||||
/** @var string|null */
|
||||
protected $queueId;
|
||||
|
||||
/** @var string */
|
||||
protected $priority = 'normal'; // 'high', 'normal', 'low'
|
||||
|
||||
/** @var array */
|
||||
protected $metadata = [];
|
||||
|
||||
/** @var array */
|
||||
protected $tags = [];
|
||||
|
||||
/** @var callable|null */
|
||||
protected $onSuccess;
|
||||
|
||||
/** @var callable|null */
|
||||
protected $onFailure;
|
||||
|
||||
/** @var callable|null */
|
||||
protected $onRetry;
|
||||
|
||||
/**
|
||||
* Set maximum retry attempts
|
||||
*
|
||||
* @param int $attempts
|
||||
* @return self
|
||||
*/
|
||||
public function maxAttempts(int $attempts): self
|
||||
{
|
||||
$this->maxAttempts = $attempts;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get maximum retry attempts
|
||||
*
|
||||
* @return int
|
||||
*/
|
||||
public function getMaxAttempts(): int
|
||||
{
|
||||
return $this->maxAttempts;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set retry delay
|
||||
*
|
||||
* @param int $seconds
|
||||
* @param string $strategy 'linear' or 'exponential'
|
||||
* @return self
|
||||
*/
|
||||
public function retryDelay(int $seconds, string $strategy = 'exponential'): self
|
||||
{
|
||||
$this->retryDelay = $seconds;
|
||||
$this->retryStrategy = $strategy;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current retry count
|
||||
*
|
||||
* @return int
|
||||
*/
|
||||
public function getRetryCount(): int
|
||||
{
|
||||
return $this->retryCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set job timeout
|
||||
*
|
||||
* @param int $seconds
|
||||
* @return self
|
||||
*/
|
||||
public function timeout(int $seconds): self
|
||||
{
|
||||
$this->timeout = $seconds;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set job priority
|
||||
*
|
||||
* @param string $priority 'high', 'normal', or 'low'
|
||||
* @return self
|
||||
*/
|
||||
public function priority(string $priority): self
|
||||
{
|
||||
if (!in_array($priority, ['high', 'normal', 'low'])) {
|
||||
throw new \InvalidArgumentException('Priority must be high, normal, or low');
|
||||
}
|
||||
$this->priority = $priority;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get job priority
|
||||
*
|
||||
* @return string
|
||||
*/
|
||||
public function getPriority(): string
|
||||
{
|
||||
return $this->priority;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add job dependency
|
||||
*
|
||||
* @param string $jobId
|
||||
* @return self
|
||||
*/
|
||||
public function dependsOn(string $jobId): self
|
||||
{
|
||||
$this->dependencies[] = $jobId;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Chain another job to run after this one
|
||||
*
|
||||
* @param Job $job
|
||||
* @param bool $onlyOnSuccess Run only if current job succeeds
|
||||
* @return self
|
||||
*/
|
||||
public function chain(Job $job, bool $onlyOnSuccess = true): self
|
||||
{
|
||||
$this->chainedJobs[] = [
|
||||
'job' => $job,
|
||||
'onlyOnSuccess' => $onlyOnSuccess,
|
||||
];
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add metadata to the job
|
||||
*
|
||||
* @param string $key
|
||||
* @param mixed $value
|
||||
* @return self
|
||||
*/
|
||||
public function withMetadata(string $key, $value): self
|
||||
{
|
||||
$this->metadata[$key] = $value;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add tags to the job
|
||||
*
|
||||
* @param array $tags
|
||||
* @return self
|
||||
*/
|
||||
public function withTags(array $tags): self
|
||||
{
|
||||
$this->tags = array_merge($this->tags, $tags);
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set success callback
|
||||
*
|
||||
* @param callable $callback
|
||||
* @return self
|
||||
*/
|
||||
public function onSuccess(callable $callback): self
|
||||
{
|
||||
$this->onSuccess = $callback;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set failure callback
|
||||
*
|
||||
* @param callable $callback
|
||||
* @return self
|
||||
*/
|
||||
public function onFailure(callable $callback): self
|
||||
{
|
||||
$this->onFailure = $callback;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set retry callback
|
||||
*
|
||||
* @param callable $callback
|
||||
* @return self
|
||||
*/
|
||||
public function onRetry(callable $callback): self
|
||||
{
|
||||
$this->onRetry = $callback;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the job with retry support
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
public function runWithRetry(): bool
|
||||
{
|
||||
$attempts = 0;
|
||||
$lastException = null;
|
||||
|
||||
while ($attempts < $this->maxAttempts) {
|
||||
$attempts++;
|
||||
$this->retryCount = $attempts - 1;
|
||||
|
||||
try {
|
||||
// Record execution start time
|
||||
$this->executionStartTime = microtime(true);
|
||||
|
||||
// Run the job
|
||||
$result = $this->run();
|
||||
|
||||
// Record execution time
|
||||
$this->executionTime = microtime(true) - $this->executionStartTime;
|
||||
|
||||
if ($result && $this->isSuccessful()) {
|
||||
// Call success callback
|
||||
if ($this->onSuccess) {
|
||||
call_user_func($this->onSuccess, $this);
|
||||
}
|
||||
|
||||
// Run chained jobs
|
||||
$this->runChainedJobs(true);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
throw new RuntimeException('Job execution failed');
|
||||
|
||||
} catch (Exception $e) {
|
||||
$lastException = $e;
|
||||
$this->output = $e->getMessage();
|
||||
$this->successful = false;
|
||||
|
||||
if ($attempts < $this->maxAttempts) {
|
||||
// Call retry callback
|
||||
if ($this->onRetry) {
|
||||
call_user_func($this->onRetry, $this, $attempts, $e);
|
||||
}
|
||||
|
||||
// Calculate delay before retry
|
||||
$delay = $this->calculateRetryDelay($attempts);
|
||||
if ($delay > 0) {
|
||||
sleep($delay);
|
||||
}
|
||||
} else {
|
||||
// Final failure
|
||||
if ($this->onFailure) {
|
||||
call_user_func($this->onFailure, $this, $e);
|
||||
}
|
||||
|
||||
// Run chained jobs that should run on failure
|
||||
$this->runChainedJobs(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Override parent run method to add timeout support
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
public function run(): bool
|
||||
{
|
||||
// Check dependencies
|
||||
if (!$this->checkDependencies()) {
|
||||
$this->output = 'Dependencies not met';
|
||||
$this->successful = false;
|
||||
return false;
|
||||
}
|
||||
|
||||
// Call parent run method
|
||||
$result = parent::run();
|
||||
|
||||
// Apply timeout to process if applicable
|
||||
if ($this->process instanceof Process && $this->timeout > 0) {
|
||||
$this->process->setTimeout($this->timeout);
|
||||
}
|
||||
|
||||
return $result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get execution time in seconds
|
||||
*
|
||||
* @return float
|
||||
*/
|
||||
public function getExecutionTime(): float
|
||||
{
|
||||
return $this->executionTime;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get job metadata
|
||||
*
|
||||
* @param string|null $key
|
||||
* @return mixed
|
||||
*/
|
||||
public function getMetadata(string $key = null)
|
||||
{
|
||||
if ($key === null) {
|
||||
return $this->metadata;
|
||||
}
|
||||
|
||||
return $this->metadata[$key] ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get job tags
|
||||
*
|
||||
* @return array
|
||||
*/
|
||||
public function getTags(): array
|
||||
{
|
||||
return $this->tags;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if job has a specific tag
|
||||
*
|
||||
* @param string $tag
|
||||
* @return bool
|
||||
*/
|
||||
public function hasTag(string $tag): bool
|
||||
{
|
||||
return in_array($tag, $this->tags);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set queue ID
|
||||
*
|
||||
* @param string $queueId
|
||||
* @return self
|
||||
*/
|
||||
public function setQueueId(string $queueId): self
|
||||
{
|
||||
$this->queueId = $queueId;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get queue ID
|
||||
*
|
||||
* @return string|null
|
||||
*/
|
||||
public function getQueueId(): ?string
|
||||
{
|
||||
return $this->queueId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get process (for background jobs)
|
||||
*
|
||||
* @return Process|null
|
||||
*/
|
||||
public function getProcess(): ?Process
|
||||
{
|
||||
return $this->process;
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculate retry delay based on strategy
|
||||
*
|
||||
* @param int $attempt
|
||||
* @return int
|
||||
*/
|
||||
protected function calculateRetryDelay(int $attempt): int
|
||||
{
|
||||
if ($this->retryStrategy === 'exponential') {
|
||||
return min($this->retryDelay * pow(2, $attempt - 1), 3600); // Max 1 hour
|
||||
}
|
||||
|
||||
return $this->retryDelay;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if dependencies are met
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
protected function checkDependencies(): bool
|
||||
{
|
||||
if (empty($this->dependencies)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// This would need to check against job history or status
|
||||
// For now, we'll assume dependencies are met
|
||||
// In a real implementation, this would check the ModernScheduler's job status
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run chained jobs
|
||||
*
|
||||
* @param bool $success Whether the current job succeeded
|
||||
* @return void
|
||||
*/
|
||||
protected function runChainedJobs(bool $success): void
|
||||
{
|
||||
foreach ($this->chainedJobs as $chainedJob) {
|
||||
$shouldRun = !$chainedJob['onlyOnSuccess'] || $success;
|
||||
|
||||
if ($shouldRun) {
|
||||
$job = $chainedJob['job'];
|
||||
if ($job instanceof ModernJob) {
|
||||
$job->runWithRetry();
|
||||
} else {
|
||||
$job->run();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert job to array for serialization
|
||||
*
|
||||
* @return array
|
||||
*/
|
||||
public function toArray(): array
|
||||
{
|
||||
return [
|
||||
'id' => $this->getId(),
|
||||
'command' => is_string($this->command) ? $this->command : 'Closure',
|
||||
'at' => $this->getAt(),
|
||||
'enabled' => $this->getEnabled(),
|
||||
'priority' => $this->priority,
|
||||
'max_attempts' => $this->maxAttempts,
|
||||
'retry_count' => $this->retryCount,
|
||||
'retry_delay' => $this->retryDelay,
|
||||
'retry_strategy' => $this->retryStrategy,
|
||||
'timeout' => $this->timeout,
|
||||
'dependencies' => $this->dependencies,
|
||||
'metadata' => $this->metadata,
|
||||
'tags' => $this->tags,
|
||||
'execution_time' => $this->executionTime,
|
||||
'successful' => $this->successful,
|
||||
'output' => $this->output,
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* Create job from array
|
||||
*
|
||||
* @param array $data
|
||||
* @return self
|
||||
*/
|
||||
public static function fromArray(array $data): self
|
||||
{
|
||||
$job = new self($data['command'] ?? '', [], $data['id'] ?? null);
|
||||
|
||||
if (isset($data['at'])) {
|
||||
$job->at($data['at']);
|
||||
}
|
||||
|
||||
if (isset($data['priority'])) {
|
||||
$job->priority($data['priority']);
|
||||
}
|
||||
|
||||
if (isset($data['max_attempts'])) {
|
||||
$job->maxAttempts($data['max_attempts']);
|
||||
}
|
||||
|
||||
if (isset($data['retry_delay']) && isset($data['retry_strategy'])) {
|
||||
$job->retryDelay($data['retry_delay'], $data['retry_strategy']);
|
||||
}
|
||||
|
||||
if (isset($data['timeout'])) {
|
||||
$job->timeout($data['timeout']);
|
||||
}
|
||||
|
||||
if (isset($data['dependencies'])) {
|
||||
foreach ($data['dependencies'] as $dep) {
|
||||
$job->dependsOn($dep);
|
||||
}
|
||||
}
|
||||
|
||||
if (isset($data['metadata'])) {
|
||||
foreach ($data['metadata'] as $key => $value) {
|
||||
$job->withMetadata($key, $value);
|
||||
}
|
||||
}
|
||||
|
||||
if (isset($data['tags'])) {
|
||||
$job->withTags($data['tags']);
|
||||
}
|
||||
|
||||
return $job;
|
||||
}
|
||||
}
|
||||
@@ -1,687 +0,0 @@
|
||||
<?php
|
||||
|
||||
/**
|
||||
* @package Grav\Common\Scheduler
|
||||
*
|
||||
* @copyright Copyright (c) 2015 - 2025 Trilby Media, LLC. All rights reserved.
|
||||
* @license MIT License; see LICENSE file for details.
|
||||
*/
|
||||
|
||||
namespace Grav\Common\Scheduler;
|
||||
|
||||
use DateTime;
|
||||
use Grav\Common\Filesystem\Folder;
|
||||
use Grav\Common\Grav;
|
||||
use Grav\Common\Utils;
|
||||
use RocketTheme\Toolbox\File\YamlFile;
|
||||
use RocketTheme\Toolbox\File\JsonFile;
|
||||
use Symfony\Component\Process\Process;
|
||||
use InvalidArgumentException;
|
||||
use RuntimeException;
|
||||
|
||||
/**
|
||||
* Modern Scheduler with enhanced features for reliability and monitoring
|
||||
*
|
||||
* @package Grav\Common\Scheduler
|
||||
*/
|
||||
class ModernScheduler extends Scheduler
|
||||
{
|
||||
/** @var array */
|
||||
protected $workers = [];
|
||||
|
||||
/** @var int */
|
||||
protected $maxWorkers = 1;
|
||||
|
||||
/** @var string */
|
||||
protected $queuePath;
|
||||
|
||||
/** @var string */
|
||||
protected $historyPath;
|
||||
|
||||
/** @var array */
|
||||
protected $modernConfig;
|
||||
|
||||
/** @var bool */
|
||||
protected $webhookEnabled = false;
|
||||
|
||||
/** @var string|null */
|
||||
protected $webhookToken;
|
||||
|
||||
/** @var bool */
|
||||
protected $healthEnabled = true;
|
||||
|
||||
/** @var JobQueue */
|
||||
protected $jobQueue;
|
||||
|
||||
/**
|
||||
* Create new ModernScheduler instance
|
||||
*/
|
||||
public function __construct()
|
||||
{
|
||||
parent::__construct();
|
||||
|
||||
$grav = Grav::instance();
|
||||
$this->modernConfig = $grav['config']->get('scheduler.modern', []);
|
||||
|
||||
// Set up modern features if enabled
|
||||
if ($this->isModernEnabled()) {
|
||||
$this->initializeModernFeatures();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if modern features are enabled
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
public function isModernEnabled(): bool
|
||||
{
|
||||
return $this->modernConfig['enabled'] ?? false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Override to create ModernJob instances
|
||||
*
|
||||
* @param callable $fn
|
||||
* @param array $args
|
||||
* @param string|null $id
|
||||
* @return ModernJob
|
||||
*/
|
||||
public function addFunction(callable $fn, $args = [], $id = null): Job
|
||||
{
|
||||
$job = new ModernJob($fn, $args, $id);
|
||||
$this->queueJob($job->configure($this->config));
|
||||
|
||||
return $job;
|
||||
}
|
||||
|
||||
/**
|
||||
* Override to create ModernJob instances
|
||||
*
|
||||
* @param string $command
|
||||
* @param array $args
|
||||
* @param string|null $id
|
||||
* @return ModernJob
|
||||
*/
|
||||
public function addCommand($command, $args = [], $id = null): Job
|
||||
{
|
||||
$job = new ModernJob($command, $args, $id);
|
||||
$this->queueJob($job->configure($this->config));
|
||||
|
||||
return $job;
|
||||
}
|
||||
|
||||
/**
|
||||
* Override to create ModernJob instances
|
||||
*
|
||||
* @param string $command
|
||||
* @param array $args
|
||||
* @param string|null $id
|
||||
* @return ModernJob
|
||||
*/
|
||||
public function raw($command, $args = [], $id = null): Job
|
||||
{
|
||||
return $this->addCommand($command, $args, $id);
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize modern scheduler features
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
protected function initializeModernFeatures(): void
|
||||
{
|
||||
$locator = Grav::instance()['locator'];
|
||||
|
||||
// Set up paths
|
||||
$this->queuePath = $this->modernConfig['queue']['path'] ?? 'user-data://scheduler/queue';
|
||||
$this->queuePath = $locator->findResource($this->queuePath, true, true);
|
||||
|
||||
$this->historyPath = $this->modernConfig['history']['path'] ?? 'user-data://scheduler/history';
|
||||
$this->historyPath = $locator->findResource($this->historyPath, true, true);
|
||||
|
||||
// Create directories if they don't exist
|
||||
if (!file_exists($this->queuePath)) {
|
||||
Folder::create($this->queuePath);
|
||||
}
|
||||
|
||||
if (!file_exists($this->historyPath)) {
|
||||
Folder::create($this->historyPath);
|
||||
}
|
||||
|
||||
// Initialize job queue
|
||||
$this->jobQueue = new JobQueue($this->queuePath);
|
||||
|
||||
// Configure workers
|
||||
$this->maxWorkers = $this->modernConfig['workers'] ?? 1;
|
||||
|
||||
// Configure webhook
|
||||
$this->webhookEnabled = $this->modernConfig['webhook']['enabled'] ?? false;
|
||||
$this->webhookToken = $this->modernConfig['webhook']['token'] ?? null;
|
||||
|
||||
// Configure health check
|
||||
$this->healthEnabled = $this->modernConfig['health']['enabled'] ?? true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Enhanced run method with modern features
|
||||
*
|
||||
* @param DateTime|null $runTime
|
||||
* @param bool $force
|
||||
* @return void
|
||||
*/
|
||||
public function run(DateTime $runTime = null, $force = false): void
|
||||
{
|
||||
if (!$this->isModernEnabled()) {
|
||||
// Fall back to parent implementation
|
||||
parent::run($runTime, $force);
|
||||
return;
|
||||
}
|
||||
|
||||
$this->loadSavedJobs();
|
||||
|
||||
if (null === $runTime) {
|
||||
$runTime = new DateTime('now');
|
||||
}
|
||||
|
||||
// Process queued jobs first
|
||||
$this->processQueuedJobs();
|
||||
|
||||
// Get scheduled jobs
|
||||
[$background, $foreground] = $this->getQueuedJobs(false);
|
||||
$alljobs = array_merge($background, $foreground);
|
||||
|
||||
// Check which jobs are due and add them to the queue
|
||||
foreach ($alljobs as $job) {
|
||||
if ($job->isDue($runTime) || $force) {
|
||||
if ($job instanceof ModernJob) {
|
||||
// Add to queue for processing
|
||||
$this->jobQueue->push($job);
|
||||
} else {
|
||||
// Run legacy jobs directly
|
||||
$job->run();
|
||||
$this->jobs_run[] = $job;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Process jobs with workers
|
||||
$this->processJobsWithWorkers();
|
||||
|
||||
// Store states and history
|
||||
$this->saveJobStates();
|
||||
$this->saveJobHistory();
|
||||
|
||||
// Update last run timestamp
|
||||
$this->updateLastRun();
|
||||
}
|
||||
|
||||
/**
|
||||
* Process jobs from the queue
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
protected function processQueuedJobs(): void
|
||||
{
|
||||
// Process existing queued jobs from previous runs
|
||||
while (!$this->jobQueue->isEmpty() && count($this->workers) < $this->maxWorkers) {
|
||||
$job = $this->jobQueue->pop();
|
||||
|
||||
if ($job) {
|
||||
$this->executeJob($job);
|
||||
$this->jobs_run[] = $job;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Process jobs using multiple workers
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
protected function processJobsWithWorkers(): void
|
||||
{
|
||||
// Process all queued jobs
|
||||
while (!$this->jobQueue->isEmpty()) {
|
||||
// Wait if we've reached max workers
|
||||
while (count($this->workers) >= $this->maxWorkers) {
|
||||
foreach ($this->workers as $workerId => $process) {
|
||||
if ($process instanceof Process && !$process->isRunning()) {
|
||||
unset($this->workers[$workerId]);
|
||||
}
|
||||
}
|
||||
if (count($this->workers) >= $this->maxWorkers) {
|
||||
usleep(100000); // Wait 100ms
|
||||
}
|
||||
}
|
||||
|
||||
// Get next job from queue
|
||||
$job = $this->jobQueue->pop();
|
||||
if ($job) {
|
||||
$this->executeJob($job);
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for all remaining workers to complete
|
||||
foreach ($this->workers as $workerId => $process) {
|
||||
if ($process instanceof Process) {
|
||||
$process->wait();
|
||||
unset($this->workers[$workerId]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a job with retry support
|
||||
*
|
||||
* @param Job $job
|
||||
* @return void
|
||||
*/
|
||||
protected function executeJob(Job $job): void
|
||||
{
|
||||
if ($job instanceof ModernJob) {
|
||||
// Use modern job execution with retry
|
||||
$job->runWithRetry();
|
||||
} else {
|
||||
// Use standard job execution
|
||||
$job->run();
|
||||
}
|
||||
|
||||
$this->jobs_run[] = $job;
|
||||
|
||||
// Handle background jobs
|
||||
if ($job->runInBackground() && $this->maxWorkers > 1) {
|
||||
$process = $job->getProcess();
|
||||
if ($process) {
|
||||
$this->workers[] = $process;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save job execution history
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
protected function saveJobHistory(): void
|
||||
{
|
||||
if (!$this->modernConfig['history']['enabled'] ?? true) {
|
||||
return;
|
||||
}
|
||||
|
||||
$now = new DateTime('now');
|
||||
$historyFile = $this->historyPath . '/' . $now->format('Y-m-d') . '.json';
|
||||
|
||||
$history = [];
|
||||
if (file_exists($historyFile)) {
|
||||
$file = JsonFile::instance($historyFile);
|
||||
$history = $file->content();
|
||||
} else {
|
||||
$file = JsonFile::instance($historyFile);
|
||||
}
|
||||
|
||||
foreach ($this->jobs_run as $job) {
|
||||
$entry = [
|
||||
'job_id' => $job->getId(),
|
||||
'command' => is_string($job->getCommand()) ? $job->getCommand() : 'Closure',
|
||||
'timestamp' => $now->format('c'),
|
||||
'success' => $job->isSuccessful(),
|
||||
'duration' => $job instanceof ModernJob ? $job->getExecutionTime() : null,
|
||||
'output' => substr($job->getOutput(), 0, 1000), // Limit output size
|
||||
'retry_count' => $job instanceof ModernJob ? $job->getRetryCount() : 0,
|
||||
];
|
||||
|
||||
$history[] = $entry;
|
||||
}
|
||||
|
||||
$file->save($history);
|
||||
|
||||
// Clean up old history files
|
||||
$this->cleanupHistory();
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up old history files
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
protected function cleanupHistory(): void
|
||||
{
|
||||
$retentionDays = $this->modernConfig['history']['retention_days'] ?? 30;
|
||||
$cutoffDate = new DateTime("-{$retentionDays} days");
|
||||
|
||||
$files = glob($this->historyPath . '/*.json');
|
||||
foreach ($files as $file) {
|
||||
$filename = basename($file, '.json');
|
||||
$fileDate = DateTime::createFromFormat('Y-m-d', $filename);
|
||||
|
||||
if ($fileDate && $fileDate < $cutoffDate) {
|
||||
unlink($file);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Update last run timestamp
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
protected function updateLastRun(): void
|
||||
{
|
||||
$lastRunFile = $this->status_path . '/last_run.txt';
|
||||
file_put_contents($lastRunFile, (new DateTime('now'))->format('c'), LOCK_EX);
|
||||
|
||||
// Also update the legacy location for backward compatibility
|
||||
file_put_contents('logs/lastcron.run', (new DateTime('now'))->format('Y-m-d H:i:s'), LOCK_EX);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check scheduler health
|
||||
*
|
||||
* @return array
|
||||
*/
|
||||
public function getHealthStatus(): array
|
||||
{
|
||||
$lastRunFile = $this->status_path . '/last_run.txt';
|
||||
$lastRun = file_exists($lastRunFile) ? file_get_contents($lastRunFile) : null;
|
||||
|
||||
$health = [
|
||||
'status' => 'healthy',
|
||||
'last_run' => $lastRun,
|
||||
'last_run_age' => null,
|
||||
'queue_size' => 0,
|
||||
'failed_jobs_24h' => 0,
|
||||
'scheduled_jobs' => count($this->getAllJobs()),
|
||||
'modern_features' => $this->isModernEnabled(),
|
||||
'workers' => $this->maxWorkers,
|
||||
'trigger_methods' => $this->getActiveTriggers(),
|
||||
];
|
||||
|
||||
if ($lastRun) {
|
||||
$lastRunTime = new DateTime($lastRun);
|
||||
$now = new DateTime('now');
|
||||
$diff = $now->getTimestamp() - $lastRunTime->getTimestamp();
|
||||
$health['last_run_age'] = $diff;
|
||||
|
||||
// Mark as unhealthy if no run in last 10 minutes
|
||||
if ($diff > 600) {
|
||||
$health['status'] = 'warning';
|
||||
}
|
||||
|
||||
// Mark as critical if no run in last hour
|
||||
if ($diff > 3600) {
|
||||
$health['status'] = 'critical';
|
||||
}
|
||||
} else {
|
||||
$health['status'] = 'unknown';
|
||||
}
|
||||
|
||||
// Get queue size if modern features enabled
|
||||
if ($this->isModernEnabled() && $this->jobQueue) {
|
||||
$health['queue_size'] = $this->jobQueue->size();
|
||||
}
|
||||
|
||||
// Count failed jobs in last 24 hours
|
||||
$health['failed_jobs_24h'] = $this->countRecentFailures();
|
||||
|
||||
return $health;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if webhook is enabled
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
public function isWebhookEnabled(): bool
|
||||
{
|
||||
return $this->webhookEnabled;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get active trigger methods
|
||||
*
|
||||
* @return array
|
||||
*/
|
||||
public function getActiveTriggers(): array
|
||||
{
|
||||
$triggers = [];
|
||||
|
||||
// Check cron
|
||||
$cronStatus = $this->isCrontabSetup();
|
||||
if ($cronStatus === 1) {
|
||||
$triggers[] = 'cron';
|
||||
}
|
||||
|
||||
// Check systemd timer
|
||||
if ($this->isSystemdTimerActive()) {
|
||||
$triggers[] = 'systemd';
|
||||
}
|
||||
|
||||
// Check webhook
|
||||
if ($this->webhookEnabled) {
|
||||
$triggers[] = 'webhook';
|
||||
}
|
||||
|
||||
// Check for external triggers
|
||||
$lastRunFile = $this->status_path . '/last_run.txt';
|
||||
if (file_exists($lastRunFile)) {
|
||||
$lastRun = file_get_contents($lastRunFile);
|
||||
$lastRunTime = new DateTime($lastRun);
|
||||
$now = new DateTime('now');
|
||||
$diff = $now->getTimestamp() - $lastRunTime->getTimestamp();
|
||||
|
||||
if ($diff < 120) {
|
||||
$triggers[] = 'external';
|
||||
}
|
||||
}
|
||||
|
||||
return $triggers;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if systemd timer is active
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
protected function isSystemdTimerActive(): bool
|
||||
{
|
||||
if (Utils::isWindows()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
$process = new Process(['systemctl', 'is-active', 'grav-scheduler.timer']);
|
||||
$process->run();
|
||||
|
||||
return $process->isSuccessful() && trim($process->getOutput()) === 'active';
|
||||
}
|
||||
|
||||
/**
|
||||
* Count recent job failures
|
||||
*
|
||||
* @return int
|
||||
*/
|
||||
protected function countRecentFailures(): int
|
||||
{
|
||||
$count = 0;
|
||||
$cutoff = new DateTime('-24 hours');
|
||||
|
||||
// Check today's history
|
||||
$todayFile = $this->historyPath . '/' . date('Y-m-d') . '.json';
|
||||
if (file_exists($todayFile)) {
|
||||
$file = JsonFile::instance($todayFile);
|
||||
$history = $file->content();
|
||||
|
||||
foreach ($history as $entry) {
|
||||
$entryTime = new DateTime($entry['timestamp']);
|
||||
if ($entryTime > $cutoff && !$entry['success']) {
|
||||
$count++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check yesterday's history
|
||||
$yesterdayFile = $this->historyPath . '/' . $cutoff->format('Y-m-d') . '.json';
|
||||
if (file_exists($yesterdayFile)) {
|
||||
$file = JsonFile::instance($yesterdayFile);
|
||||
$history = $file->content();
|
||||
|
||||
foreach ($history as $entry) {
|
||||
$entryTime = new DateTime($entry['timestamp']);
|
||||
if ($entryTime > $cutoff && !$entry['success']) {
|
||||
$count++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return $count;
|
||||
}
|
||||
|
||||
/**
|
||||
* Process webhook trigger
|
||||
*
|
||||
* @param string|null $token
|
||||
* @param string|null $jobId
|
||||
* @return array
|
||||
*/
|
||||
public function processWebhookTrigger($token = null, $jobId = null): array
|
||||
{
|
||||
if (!$this->webhookEnabled) {
|
||||
return ['success' => false, 'message' => 'Webhook triggers are not enabled'];
|
||||
}
|
||||
|
||||
if ($this->webhookToken && $token !== $this->webhookToken) {
|
||||
return ['success' => false, 'message' => 'Invalid webhook token'];
|
||||
}
|
||||
|
||||
if ($jobId) {
|
||||
// Force run specific job (manual override - ignore schedule)
|
||||
$job = $this->getJob($jobId);
|
||||
if ($job) {
|
||||
// Force run in foreground to get immediate result
|
||||
$job->inForeground()->run();
|
||||
|
||||
// Track as manually executed
|
||||
$this->jobs_run[] = $job;
|
||||
$this->saveJobStates();
|
||||
$this->updateLastRun();
|
||||
|
||||
return [
|
||||
'success' => $job->isSuccessful(),
|
||||
'message' => $job->isSuccessful() ? 'Job force-executed successfully' : 'Job execution failed',
|
||||
'job_id' => $jobId,
|
||||
'forced' => true,
|
||||
'output' => $job->getOutput(),
|
||||
];
|
||||
} else {
|
||||
return ['success' => false, 'message' => 'Job not found: ' . $jobId];
|
||||
}
|
||||
} else {
|
||||
// Run all due jobs normally
|
||||
$this->run();
|
||||
return [
|
||||
'success' => true,
|
||||
'message' => 'Scheduler executed (due jobs only)',
|
||||
'jobs_run' => count($this->jobs_run),
|
||||
'timestamp' => date('c'),
|
||||
];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get scheduler statistics
|
||||
*
|
||||
* @return array
|
||||
*/
|
||||
public function getStatistics(): array
|
||||
{
|
||||
$stats = [
|
||||
'total_jobs' => count($this->getAllJobs()),
|
||||
'enabled_jobs' => 0,
|
||||
'disabled_jobs' => 0,
|
||||
'executions_today' => 0,
|
||||
'failures_today' => 0,
|
||||
'average_execution_time' => 0,
|
||||
'queue_size' => 0,
|
||||
];
|
||||
|
||||
// Count enabled/disabled jobs
|
||||
foreach ($this->getAllJobs() as $job) {
|
||||
if ($job->getEnabled()) {
|
||||
$stats['enabled_jobs']++;
|
||||
} else {
|
||||
$stats['disabled_jobs']++;
|
||||
}
|
||||
}
|
||||
|
||||
// Get today's statistics
|
||||
$todayFile = $this->historyPath . '/' . date('Y-m-d') . '.json';
|
||||
if (file_exists($todayFile)) {
|
||||
$file = JsonFile::instance($todayFile);
|
||||
$history = $file->content();
|
||||
|
||||
$totalTime = 0;
|
||||
$timeCount = 0;
|
||||
|
||||
foreach ($history as $entry) {
|
||||
$stats['executions_today']++;
|
||||
|
||||
if (!$entry['success']) {
|
||||
$stats['failures_today']++;
|
||||
}
|
||||
|
||||
if (isset($entry['duration']) && $entry['duration'] > 0) {
|
||||
$totalTime += $entry['duration'];
|
||||
$timeCount++;
|
||||
}
|
||||
}
|
||||
|
||||
if ($timeCount > 0) {
|
||||
$stats['average_execution_time'] = round($totalTime / $timeCount, 2);
|
||||
}
|
||||
}
|
||||
|
||||
// Get queue size
|
||||
if ($this->isModernEnabled() && $this->jobQueue) {
|
||||
$stats['queue_size'] = $this->jobQueue->size();
|
||||
}
|
||||
|
||||
return $stats;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run scheduler in daemon mode
|
||||
*
|
||||
* @param int $interval Check interval in seconds (default: 60)
|
||||
* @return void
|
||||
*/
|
||||
public function runDaemon($interval = 60): void
|
||||
{
|
||||
if (!$this->isModernEnabled()) {
|
||||
throw new RuntimeException('Daemon mode requires modern features to be enabled');
|
||||
}
|
||||
|
||||
$lastRun = 0;
|
||||
|
||||
while (true) {
|
||||
$now = time();
|
||||
|
||||
// Run scheduler every minute
|
||||
if ($now - $lastRun >= $interval) {
|
||||
$this->run();
|
||||
$lastRun = $now;
|
||||
}
|
||||
|
||||
// Process any queued jobs
|
||||
$this->processQueuedJobs();
|
||||
|
||||
// Sleep for a short interval
|
||||
sleep(5);
|
||||
|
||||
// Check for shutdown signal
|
||||
if (file_exists($this->status_path . '/shutdown')) {
|
||||
unlink($this->status_path . '/shutdown');
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -76,6 +76,9 @@ class Scheduler
|
||||
/** @var string */
|
||||
protected $historyPath;
|
||||
|
||||
/** @var JobHistory|null */
|
||||
protected $jobHistory = null;
|
||||
|
||||
/** @var array */
|
||||
protected $modernConfig = [];
|
||||
|
||||
@@ -474,6 +477,10 @@ class Scheduler
|
||||
// Initialize job queue
|
||||
$this->jobQueue = new JobQueue($this->queuePath);
|
||||
|
||||
// Initialize job history
|
||||
$retentionDays = $this->modernConfig['history']['retention_days'] ?? 30;
|
||||
$this->jobHistory = new JobHistory($this->historyPath, $retentionDays);
|
||||
|
||||
// Configure workers
|
||||
$this->maxWorkers = $this->modernConfig['workers'] ?? 1;
|
||||
|
||||
@@ -495,6 +502,16 @@ class Scheduler
|
||||
return $this->jobQueue;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the job history
|
||||
*
|
||||
* @return JobHistory|null
|
||||
*/
|
||||
public function getHistory(): ?JobHistory
|
||||
{
|
||||
return $this->jobHistory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if webhook is enabled
|
||||
*
|
||||
@@ -660,6 +677,15 @@ class Scheduler
|
||||
if (isset($worker['job'])) {
|
||||
$worker['job']->finalize();
|
||||
$this->saveJobState($worker['job']);
|
||||
|
||||
// Log background job completion to history
|
||||
if ($this->jobHistory) {
|
||||
$metadata = [
|
||||
'queue_id' => $worker['queueId'] ?? null,
|
||||
'background' => true
|
||||
];
|
||||
$this->jobHistory->logExecution($worker['job'], $metadata);
|
||||
}
|
||||
}
|
||||
|
||||
// Update queue status for background jobs
|
||||
@@ -747,6 +773,12 @@ class Scheduler
|
||||
$this->jobQueue->fail($queueId, $job->getOutput() ?: 'Job execution failed');
|
||||
}
|
||||
}
|
||||
|
||||
// Log foreground jobs immediately
|
||||
if (!$job->runInBackground() && $this->jobHistory) {
|
||||
$metadata = ['queue_id' => $queueId];
|
||||
$this->jobHistory->logExecution($job, $metadata);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user