diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index 706d17cc..0c6dbbaf 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -14,6 +14,17 @@ jobs: test: runs-on: ubuntu-latest + services: + redis: + image: redis:7-alpine + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + steps: - name: Checkout code uses: actions/checkout@v4 diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b1bd9a9..b6b3451c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 # Release Notes for 0.9.x +## [v0.10.1 (2026-10-07)](https://github.com/phenixphp/framework/compare/0.10.0...0.10.1) + +### Changed + +- Expose public URL and listen custom host. ([#138](https://github.com/phenixphp/framework/pull/138)) +- Improve Redis implementation in queues. ([#139](https://github.com/phenixphp/framework/pull/139)) + +### Removed + +- Timers in the scheduler. ([#139](https://github.com/phenixphp/framework/pull/139)) + +## [v0.10.0 (2026-06-05)](https://github.com/phenixphp/framework/compare/0.9.2...0.10.0) + +### Added + +- Support for Server-Sent Events. ([#137](https://github.com/phenixphp/framework/pull/137)) + + +## [v0.9.2 (2026-05-25)](https://github.com/phenixphp/framework/compare/0.9.1...0.9.2) + +### Fixed + +- Escape inline values from template section. ([#136](https://github.com/phenixphp/framework/pull/136)) + ## [v0.9.1 (2026-05-25)](https://github.com/phenixphp/framework/compare/0.9.0...0.9.1) ### Changed diff --git a/src/App.php b/src/App.php index 3460294d..bf73ef1e 100644 --- a/src/App.php +++ b/src/App.php @@ -40,7 +40,6 @@ use Phenix\Http\ExceptionHandler as AppExceptionHandler; use Phenix\Logging\LoggerFactory; use Phenix\Runtime\Log; -use Phenix\Scheduling\TimerRegistry; use Phenix\Session\SessionMiddlewareFactory; use function Amp\async; @@ -130,8 +129,6 @@ public function run(): void $this->isRunning = true; - TimerRegistry::run(); - if ($this->serverMode === ServerMode::CLUSTER && $this->signalTrapping) { async(function (): void { Cluster::awaitTermination(); diff --git a/src/Facades/Schedule.php b/src/Facades/Schedule.php index 64218c62..12668312 100644 --- a/src/Facades/Schedule.php +++ b/src/Facades/Schedule.php @@ -6,11 +6,9 @@ use Phenix\Runtime\Facade; use Phenix\Scheduling\Scheduler; -use Phenix\Scheduling\Timer; /** - * @method static Timer timer(Closure $closure) - * @method static Scheduler call(Closure $closure) + * @method static Scheduler call(string $name, Closure $closure) * @method static void run() * * @see \Phenix\Scheduling\Schedule diff --git a/src/Queue/LuaScripts.php b/src/Queue/LuaScripts.php index 95a48ac5..8734f7d6 100644 --- a/src/Queue/LuaScripts.php +++ b/src/Queue/LuaScripts.php @@ -4,139 +4,97 @@ namespace Phenix\Queue; -class LuaScripts +final class LuaScripts { - public static function pop(): string + public static function push(): string { return <<<'LUA' --- KEYS[1] - queue key (e.g., "queues:default") --- KEYS[2] - failed tasks set key (e.g., "queues:failed") --- KEYS[3] - delayed queue key (e.g., "queues:delayed") --- ARGV[1] - current timestamp --- ARGV[2] - reservation timeout (seconds) - --- First, move any delayed tasks that are ready back to the main queue -local current_time = tonumber(ARGV[1]) -local ready_tasks = redis.call('ZRANGEBYSCORE', KEYS[3], 0, current_time) - -if #ready_tasks > 0 then - -- Move ready tasks from delayed queue to main queue - for i = 1, #ready_tasks do - redis.call('LPUSH', KEYS[1], ready_tasks[i]) - end - -- Remove the tasks from delayed queue - redis.call('ZREMRANGEBYSCORE', KEYS[3], 0, current_time) +if redis.call('EXISTS', KEYS[2]) == 1 then + return 0 end -local function pop_next_task() - local payload = redis.call('LPOP', KEYS[1]) - if not payload then - return nil - end +redis.call('HSET', KEYS[2], + 'payload', ARGV[2], + 'queue_name', ARGV[3], + 'attempts', 0, + 'created_at', ARGV[4] +) +redis.call('RPUSH', KEYS[1], ARGV[1]) - -- Deserialize to get task ID (we need to extract it from the payload) - -- Since we can't unserialize in Lua, we'll use a pattern to extract task ID - local task_id = string.match(payload, '"taskId";s:%d+:"([^"]+)"') - if not task_id then - -- If we can't extract task ID, assume it's not failed and return it - return payload - end +return 1 +LUA; + } - -- Check if task is in failed set - local failed_key = "task:failed:" .. task_id - if redis.call('EXISTS', failed_key) == 1 then - -- Task has failed, skip it and try next - return nil - end + public static function pop(): string + { + return <<<'LUA' +local now = tonumber(ARGV[1]) +local reserved_until = tonumber(ARGV[2]) +local receipt = ARGV[3] + +local expired = redis.call('ZRANGEBYSCORE', KEYS[2], 0, now, 'LIMIT', 0, 100) +for _, task_id in ipairs(expired) do + redis.call('ZREM', KEYS[2], task_id) + redis.call('RPUSH', KEYS[1], task_id) + redis.call('HDEL', 'task:data:' .. task_id, 'reserved_at', 'reserved_until', 'receipt') +end - return payload +local delayed = redis.call('ZRANGEBYSCORE', KEYS[3], 0, now, 'LIMIT', 0, 100) +for _, task_id in ipairs(delayed) do + redis.call('ZREM', KEYS[3], task_id) + redis.call('RPUSH', KEYS[1], task_id) end --- Try to pop a valid task (not failed) -local max_attempts = 10 -local attempts = 0 +while true do + local task_id = redis.call('LPOP', KEYS[1]) + if not task_id then + return nil + end -while attempts < max_attempts do - local payload = pop_next_task() + local data_key = 'task:data:' .. task_id + local payload = redis.call('HGET', data_key, 'payload') if payload then - return payload + local attempts = redis.call('HINCRBY', data_key, 'attempts', 1) + redis.call('HSET', data_key, 'reserved_at', now, 'reserved_until', reserved_until, 'receipt', receipt) + redis.call('ZADD', KEYS[2], reserved_until, task_id) + return {task_id, payload, attempts} end - attempts = attempts + 1 end - --- No valid task found after max attempts -return nil LUA; } - public static function reserve(): string + public static function complete(): string { return <<<'LUA' --- KEYS[1] - reserved key (e.g., "task:reserved:task_id") --- KEYS[2] - task data key (e.g., "task:data:task_id") --- ARGV[1] - expiration timestamp --- ARGV[2] - attempts count --- ARGV[3] - reserved_at timestamp --- ARGV[4] - reserved_until timestamp --- ARGV[5] - payload --- ARGV[6] - expire_in_seconds - --- Try to set reservation (atomic) -if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then - -- Successfully reserved, now set task data - redis.call('HSET', KEYS[2], - 'attempts', ARGV[2], - 'reserved_at', ARGV[3], - 'reserved_until', ARGV[4], - 'payload', ARGV[5] - ) - - -- Set expiration for task data - redis.call('EXPIRE', KEYS[2], ARGV[6]) - - return 1 -else +if redis.call('HGET', KEYS[2], 'receipt') ~= ARGV[2] then return 0 end -LUA; - } - public static function push(): string - { - return <<<'LUA' --- KEYS[1] - queue key (e.g., "queues:default") --- ARGV[1] - serialized task payload - -return redis.call('RPUSH', KEYS[1], ARGV[1]) +redis.call('ZREM', KEYS[1], ARGV[1]) +redis.call('DEL', KEYS[2]) +return 1 LUA; } public static function retry(): string { return <<<'LUA' --- KEYS[1] - reserved key (e.g., "task:reserved:task_id") --- KEYS[2] - task data key (e.g., "task:data:task_id") --- KEYS[3] - queue key (e.g., "queues:default") --- KEYS[4] - delayed queue key (e.g., "queues:delayed") --- ARGV[1] - attempts count --- ARGV[2] - payload --- ARGV[3] - delay (0 for immediate, >0 for delayed) --- ARGV[4] - execute_at timestamp (only used if delay > 0) - --- Remove reservation -redis.call('DEL', KEYS[1]) +if redis.call('HGET', KEYS[2], 'receipt') ~= ARGV[5] then + return 0 +end --- Update task data with new attempt count -redis.call('HSET', KEYS[2], 'attempts', ARGV[1]) +redis.call('ZREM', KEYS[1], ARGV[1]) +redis.call('HSET', KEYS[2], 'payload', ARGV[2]) +redis.call('HDEL', KEYS[2], 'reserved_at', 'reserved_until', 'receipt') if tonumber(ARGV[3]) > 0 then - -- Add to delayed queue with score = execute_at timestamp - redis.call('ZADD', KEYS[4], ARGV[4], ARGV[2]) + redis.call('ZADD', KEYS[4], ARGV[4], ARGV[1]) else - -- Add back to main queue immediately - redis.call('RPUSH', KEYS[3], ARGV[2]) + redis.call('RPUSH', KEYS[3], ARGV[1]) end +redis.call('DEL', KEYS[5]) +redis.call('LREM', KEYS[6], 0, ARGV[1]) return 1 LUA; } @@ -144,67 +102,44 @@ public static function retry(): string public static function fail(): string { return <<<'LUA' --- KEYS[1] - reserved key (e.g., "task:reserved:task_id") --- KEYS[2] - task data key (e.g., "task:data:task_id") --- KEYS[3] - failed key (e.g., "task:failed:task_id") --- KEYS[4] - failed queue (e.g., "queues:failed") --- ARGV[1] - task_id --- ARGV[2] - failed_at timestamp --- ARGV[3] - exception data (JSON) --- ARGV[4] - payload - --- Create failed task record +if redis.call('HGET', KEYS[2], 'receipt') ~= ARGV[6] then + return 0 +end + +redis.call('ZREM', KEYS[1], ARGV[1]) redis.call('HSET', KEYS[3], 'task_id', ARGV[1], 'failed_at', ARGV[2], 'exception', ARGV[3], - 'payload', ARGV[4] + 'payload', ARGV[4], + 'queue_name', ARGV[5] ) - --- Add task ID to failed queue for management redis.call('LPUSH', KEYS[4], ARGV[1]) - --- Clean up reservation and task data -redis.call('DEL', KEYS[1], KEYS[2]) - +redis.call('DEL', KEYS[2]) return 1 LUA; } - public static function cleanupExpiredReservations(): string + public static function clear(): string { return <<<'LUA' --- ARGV[1] - current timestamp - -local cursor = 0 -local cleaned_count = 0 -local batch_size = 100 - -repeat - local result = redis.call("SCAN", cursor, "MATCH", "task:reserved:*", "COUNT", batch_size) - cursor = tonumber(result[1]) - local reserved_keys = result[2] - - for i = 1, #reserved_keys do - local key = reserved_keys[i] - local expiration = redis.call("GET", key) - - if expiration and tonumber(expiration) < tonumber(ARGV[1]) then - redis.call("DEL", key) - cleaned_count = cleaned_count + 1 - - -- Extract task ID from key and update task data - local task_id = string.match(key, "task:reserved:(.+)") - if task_id then - local task_data_key = "task:data:" .. task_id - redis.call("HDEL", task_data_key, "reserved_at") - redis.call("HSET", task_data_key, "available_at", ARGV[1]) - end - end +local ids = redis.call('LRANGE', KEYS[1], 0, -1) +local delayed = redis.call('ZRANGE', KEYS[3], 0, -1) + +for _, task_id in ipairs(ids) do + if not redis.call('ZSCORE', KEYS[2], task_id) then + redis.call('DEL', 'task:data:' .. task_id) end -until cursor == 0 +end + +for _, task_id in ipairs(delayed) do + if not redis.call('ZSCORE', KEYS[2], task_id) then + redis.call('DEL', 'task:data:' .. task_id) + end +end -return cleaned_count +redis.call('DEL', KEYS[1], KEYS[3]) +return #ids + #delayed LUA; } } diff --git a/src/Queue/QueueManager.php b/src/Queue/QueueManager.php index 88d063f4..11417994 100644 --- a/src/Queue/QueueManager.php +++ b/src/Queue/QueueManager.php @@ -129,7 +129,8 @@ protected function createRedisDriver(): Queue return new RedisQueue( redis: $client, - queueName: $config['queue'] ?? 'default' + queueName: $config['queue'] ?? 'default', + reservationTimeout: (int) ($config['reservation_timeout'] ?? 60) ); } } diff --git a/src/Queue/RedisQueue.php b/src/Queue/RedisQueue.php index f0738833..731440d8 100644 --- a/src/Queue/RedisQueue.php +++ b/src/Queue/RedisQueue.php @@ -8,13 +8,15 @@ use Phenix\Redis\Contracts\Client; use Phenix\Tasks\QueuableTask; +use function is_array; use function is_int; class RedisQueue extends Queue { public function __construct( protected Client $redis, - string|null $queueName = 'default' + string|null $queueName = 'default', + protected int $reservationTimeout = 60 ) { parent::__construct($queueName); @@ -23,17 +25,27 @@ public function __construct( public function size(): int { - $result = $this->redis->execute('LLEN', $this->getQueueKey()); + $result = $this->redis->execute('LLEN', $this->readyKey()); return is_int($result) ? $result : 0; } public function push(QueuableTask $task): void { - $queueKey = $this->getQueueKey($task->getQueueName()); - $payload = $task->getPayload(); + $queue = $task->getQueueName() ?? $this->queueName ?? 'default'; + $task->setQueueName($queue); - $this->redis->execute('RPUSH', $queueKey, $payload); + $this->redis->execute( + 'EVAL', + LuaScripts::push(), + 2, + $this->readyKey($queue), + $this->taskDataKey($task->getTaskId()), + $task->getTaskId(), + $task->getPayload(), + $queue, + time() + ); } public function pushOn(string $queue, QueuableTask $task): static @@ -46,34 +58,46 @@ public function pushOn(string $queue, QueuableTask $task): static public function pop(string|null $queueName = null): QueuableTask|null { - $queueKey = $this->getQueueKey($queueName); - $failedQueueKey = "queues:failed"; - $delayedQueueKey = "queues:delayed"; - - $payload = $this->redis->execute('EVAL', LuaScripts::pop(), 3, $queueKey, $failedQueueKey, $delayedQueueKey, time(), 60); - - if ($payload === null) { + $queue = $queueName ?? $this->queueName ?? 'default'; + $now = time(); + $receipt = bin2hex(random_bytes(16)); + $result = $this->redis->execute( + 'EVAL', + LuaScripts::pop(), + 3, + $this->readyKey($queue), + $this->reservedKey($queue), + $this->delayedKey($queue), + $now, + $now + $this->reservationTimeout, + $receipt + ); + + if (! is_array($result) || count($result) !== 3) { return null; } + [$taskId, $payload, $attempts] = $result; $task = $this->restoreTask($payload); - if ($task) { - $task->setQueueName($queueName ?? $this->queueName ?? 'default'); + if ($task === null) { + $this->redis->execute('ZREM', $this->reservedKey($queue), (string) $taskId); + $this->redis->execute('DEL', $this->taskDataKey((string) $taskId)); - if ($this->stateManager->reserve($task)) { - return $task; - } + return null; + } + + $task->setTaskId((string) $taskId); + $task->setQueueName($queue); + $task->setAttempts((int) $attempts); - $this->redis->execute('RPUSH', $queueKey, $payload); + if ($this->stateManager instanceof RedisTaskState) { + $this->stateManager->registerReceipt($task, $receipt); } - return null; + return $task; } - /** - * @return array - */ public function popChunk(int $limit, string|null $queueName = null): array { if ($limit <= 0) { @@ -97,13 +121,34 @@ public function popChunk(int $limit, string|null $queueName = null): array public function clear(): void { - $this->redis->execute('DEL', $this->getQueueKey()); + $queue = $this->queueName ?? 'default'; + $this->redis->execute( + 'EVAL', + LuaScripts::clear(), + 3, + $this->readyKey($queue), + $this->reservedKey($queue), + $this->delayedKey($queue) + ); } - protected function getQueueKey(string|null $queueName = null): string + protected function readyKey(string|null $queueName = null): string { - $queue = $queueName ?? $this->queueName ?? 'default'; + return 'queues:' . ($queueName ?? $this->queueName ?? 'default') . ':ready'; + } + + protected function reservedKey(string|null $queueName = null): string + { + return 'queues:' . ($queueName ?? $this->queueName ?? 'default') . ':reserved'; + } - return "queues:{$queue}"; + protected function delayedKey(string|null $queueName = null): string + { + return 'queues:' . ($queueName ?? $this->queueName ?? 'default') . ':delayed'; + } + + protected function taskDataKey(string $taskId): string + { + return "task:data:{$taskId}"; } } diff --git a/src/Queue/StateManagers/RedisTaskState.php b/src/Queue/StateManagers/RedisTaskState.php index fd651920..f7d8a2af 100644 --- a/src/Queue/StateManagers/RedisTaskState.php +++ b/src/Queue/StateManagers/RedisTaskState.php @@ -4,139 +4,128 @@ namespace Phenix\Queue\StateManagers; +use LogicException; use Phenix\Queue\Contracts\TaskState; use Phenix\Queue\LuaScripts; use Phenix\Redis\Contracts\Client; use Phenix\Tasks\QueuableTask; use Throwable; +use WeakMap; class RedisTaskState implements TaskState { - public function __construct( - protected Client $redis - ) { + /** @var WeakMap */ + protected WeakMap $receipts; + + public function __construct(protected Client $redis) + { + $this->receipts = new WeakMap(); } public function reserve(QueuableTask $task, int $timeout = 60): bool { - $taskId = $this->getTaskId($task); - $reservedKey = "task:reserved:{$taskId}"; - $taskDataKey = "task:data:{$taskId}"; - - $reserved = $this->redis->execute('SETNX', $reservedKey, time() + $timeout); - - if ($reserved) { - $currentAttempts = $task->getAttempts(); - $newAttempts = $currentAttempts === 0 ? $currentAttempts + 1 : $currentAttempts; - - $taskData = [ - 'attempts' => $newAttempts, - 'reserved_at' => time(), - 'reserved_until' => time() + $timeout, - 'payload' => $task->getPayload(), - ]; - - $this->redis->execute('HSET', $taskDataKey, ...$this->flattenArray($taskData)); - $this->redis->execute('EXPIRE', $taskDataKey, $timeout + 300); - - if ($currentAttempts === 0) { - $task->setAttempts($newAttempts); - } - - return true; - } + return isset($this->receipts[$task]); + } - return false; + public function registerReceipt(QueuableTask $task, string $receipt): void + { + $this->receipts[$task] = $receipt; } public function complete(QueuableTask $task): void { - $taskId = $this->getTaskId($task); - $reservedKey = "task:reserved:{$taskId}"; - $taskDataKey = "task:data:{$taskId}"; + $this->redis->execute( + 'EVAL', + LuaScripts::complete(), + 2, + $this->reservedKey($task), + $this->taskDataKey($task), + $task->getTaskId(), + $this->receipt($task) + ); - $this->redis->execute('DEL', $reservedKey, $taskDataKey); + unset($this->receipts[$task]); } public function fail(QueuableTask $task, Throwable $exception): void { - $taskId = $this->getTaskId($task); - $reservedKey = "task:reserved:{$taskId}"; - $taskDataKey = "task:data:{$taskId}"; - $failedKey = "task:failed:{$taskId}"; - - $failedData = [ - 'task_id' => $taskId, - 'failed_at' => time(), - 'exception' => json_encode([ + $taskId = $task->getTaskId(); + $this->redis->execute( + 'EVAL', + LuaScripts::fail(), + 4, + $this->reservedKey($task), + $this->taskDataKey($task), + "task:failed:{$taskId}", + 'queues:failed', + $taskId, + time(), + json_encode([ 'message' => $exception->getMessage(), 'code' => $exception->getCode(), 'file' => $exception->getFile(), 'line' => $exception->getLine(), 'trace' => $exception->getTraceAsString(), ]), - 'payload' => $task->getPayload(), - ]; - - $this->redis->execute('HSET', $failedKey, ...$this->flattenArray($failedData)); - $this->redis->execute('LPUSH', 'queues:failed', $taskId); + $task->getPayload(), + $task->getQueueName() ?? 'default', + $this->receipt($task) + ); - $this->redis->execute('DEL', $reservedKey, $taskDataKey); + unset($this->receipts[$task]); } public function retry(QueuableTask $task, int $delay = 0): void { - $taskId = $this->getTaskId($task); - $reservedKey = "task:reserved:{$taskId}"; - $taskDataKey = "task:data:{$taskId}"; - $failedKey = "task:failed:{$taskId}"; - $queueKey = "queues:{$task->getQueueName()}"; - $delayedKey = "queues:delayed"; - - // Increment attempts before re-queuing for Redis-based queue - $task->setAttempts($task->getAttempts() + 1); + $taskId = $task->getTaskId(); + $queue = $task->getQueueName() ?? 'default'; $this->redis->execute( 'EVAL', LuaScripts::retry(), - 4, // number of keys - $reservedKey, - $taskDataKey, - $queueKey, - $delayedKey, - $task->getAttempts(), // ARGV[1] - now includes the incremented attempts - $task->getPayload(), // ARGV[2] - updated payload with incremented attempts - $delay, // ARGV[3] - time() + $delay // ARGV[4] + 6, + $this->reservedKey($task), + $this->taskDataKey($task), + "queues:{$queue}:ready", + "queues:{$queue}:delayed", + "task:failed:{$taskId}", + 'queues:failed', + $taskId, + $task->getPayload(), + $delay, + time() + $delay, + $this->receipt($task) ); - $this->redis->execute('DEL', $failedKey); - $this->redis->execute('LREM', 'queues:failed', 0, $taskId); + unset($this->receipts[$task]); } public function getTaskState(string $taskId): array|null { - $taskDataKey = "task:data:{$taskId}"; - $data = $this->redis->execute('HGETALL', $taskDataKey); + $data = $this->redis->execute('HGETALL', "task:data:{$taskId}"); return empty($data) ? null : $this->arrayFromRedisHash($data); } - protected function getTaskId(QueuableTask $task): string + public function cleanupExpiredReservations(): void { - return $task->getTaskId(); + // RedisQueue::pop() recovers expired reservations for the queue being consumed. } - protected function flattenArray(array $data): array + protected function taskDataKey(QueuableTask $task): string { - $flattened = []; + return 'task:data:' . $task->getTaskId(); + } - foreach ($data as $key => $value) { - $flattened[] = $key; - $flattened[] = $value; - } + protected function reservedKey(QueuableTask $task): string + { + return 'queues:' . ($task->getQueueName() ?? 'default') . ':reserved'; + } - return $flattened; + protected function receipt(QueuableTask $task): string + { + return $this->receipts[$task] + ?? throw new LogicException('Task does not have an active Redis reservation receipt.'); } protected function arrayFromRedisHash(array $hash): array @@ -149,9 +138,4 @@ protected function arrayFromRedisHash(array $hash): array return $result; } - - public function cleanupExpiredReservations(): void - { - $this->redis->execute('EVAL', LuaScripts::cleanupExpiredReservations(), 0, time()); - } } diff --git a/src/Scheduling/Contracts/ScheduleLock.php b/src/Scheduling/Contracts/ScheduleLock.php new file mode 100644 index 00000000..ac745962 --- /dev/null +++ b/src/Scheduling/Contracts/ScheduleLock.php @@ -0,0 +1,12 @@ +key($schedule, $occurrence); + $token = bin2hex(random_bytes(16)); + $script = <<<'LUA' +if redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[1], 'NX') then + return 1 +end + +return 0 +LUA; + + return $this->redis->execute('EVAL', $script, 1, $key, $ttl, $token) === 1 + ? $token + : null; + } + + public function release(string $schedule, string $occurrence, string $token): void + { + $script = <<<'LUA' +if redis.call('GET', KEYS[1]) ~= ARGV[1] then + return 0 +end + +return redis.call('DEL', KEYS[1]) +LUA; + + $this->redis->execute('EVAL', $script, 1, $this->key($schedule, $occurrence), $token); + } + + protected function key(string $schedule, string $occurrence): string + { + return $this->prefix . hash('sha256', $schedule) . ':' . $occurrence; + } +} diff --git a/src/Scheduling/Schedule.php b/src/Scheduling/Schedule.php index 3ca20366..f7b928c9 100644 --- a/src/Scheduling/Schedule.php +++ b/src/Scheduling/Schedule.php @@ -5,39 +5,34 @@ namespace Phenix\Scheduling; use Closure; +use InvalidArgumentException; +use Phenix\Scheduling\Contracts\ScheduleLock; use Phenix\Util\Date; class Schedule { - /** - * @var array - */ + /** @var array */ protected array $schedules = []; - public function timer(Closure $closure): Timer + public function __construct(protected ScheduleLock $lock, protected int $lockTtl = 86400) { - $timer = new Timer($closure); - - TimerRegistry::add($timer); - - return $timer; } - public function call(Closure $closure): Scheduler + public function call(string $name, Closure $closure): Scheduler { - $scheduler = new Scheduler($closure); - - $this->schedules[] = $scheduler; + if (isset($this->schedules[$name])) { + throw new InvalidArgumentException("Schedule [{$name}] is already registered."); + } - return $scheduler; + return $this->schedules[$name] = new Scheduler($name, $closure, $this->lock, $this->lockTtl); } public function run(): void { $now = null; + foreach ($this->schedules as $scheduler) { $now ??= Date::now('UTC'); - $scheduler->tick($now); } } diff --git a/src/Scheduling/Scheduler.php b/src/Scheduling/Scheduler.php index 862ed31a..2b7ffce4 100644 --- a/src/Scheduling/Scheduler.php +++ b/src/Scheduling/Scheduler.php @@ -6,7 +6,9 @@ use Closure; use Cron\CronExpression; +use Phenix\Scheduling\Contracts\ScheduleLock; use Phenix\Util\Date; +use Throwable; use function Amp\weakClosure; use function count; @@ -20,7 +22,10 @@ class Scheduler protected CronExpression|null $expression = null; public function __construct( - Closure $closure + protected string $name, + Closure $closure, + protected ScheduleLock $lock, + protected int $lockTtl = 86400 ) { $this->closure = weakClosure($closure); } @@ -151,8 +156,28 @@ public function tick(Date|null $now = null): void $now ??= Date::now(); $localNow = $now->copy()->timezone($this->timezone); - if ($this->expression->isDue($localNow)) { + $occurrence = $now->copy()->utc()->format('YmdHi'); + + if (! $this->expression->isDue($localNow)) { + return; + } + + $token = $this->lock->acquire($this->name, $occurrence, $this->lockTtl); + + if ($token === null) { + return; + } + + try { ($this->closure)(); + } catch (Throwable $exception) { + try { + $this->lock->release($this->name, $occurrence, $token); + } catch (Throwable) { + // Preserve the callback exception if Redis is also unavailable. + } + + throw $exception; } } } diff --git a/src/Scheduling/SchedulingServiceProvider.php b/src/Scheduling/SchedulingServiceProvider.php index e5ae6425..4738a4a1 100644 --- a/src/Scheduling/SchedulingServiceProvider.php +++ b/src/Scheduling/SchedulingServiceProvider.php @@ -4,16 +4,28 @@ namespace Phenix\Scheduling; +use Phenix\Facades\Config; use Phenix\Facades\File; +use Phenix\Facades\Redis; use Phenix\Providers\ServiceProvider; use Phenix\Scheduling\Console\ScheduleRunCommand; use Phenix\Scheduling\Console\ScheduleWorkCommand; +use Phenix\Scheduling\Contracts\ScheduleLock; class SchedulingServiceProvider extends ServiceProvider { public function boot(): void { - $this->bind(Schedule::class)->setShared(true); + $this->bind(ScheduleLock::class, fn (): ScheduleLock => new RedisScheduleLock( + Redis::connection((string) Config::get('schedule.connection', 'default'))->client(), + (string) Config::get('schedule.lock_prefix', 'phenix:schedule:') + ))->setShared(true); + + $this->bind(Schedule::class, fn (): Schedule => new Schedule( + $this->getContainer()->get(ScheduleLock::class), + (int) Config::get('schedule.occurrence_ttl', 86400) + ))->setShared(true); + $this->bind(ScheduleWorker::class); $this->commands([ diff --git a/src/Scheduling/Timer.php b/src/Scheduling/Timer.php deleted file mode 100644 index ff122b47..00000000 --- a/src/Scheduling/Timer.php +++ /dev/null @@ -1,158 +0,0 @@ -closure = weakClosure($closure); - } - - public function seconds(float $seconds): self - { - $this->interval = max(0.001, $seconds); - - return $this; - } - - public function milliseconds(int $milliseconds): self - { - $this->interval = max(0.001, $milliseconds / 1000); - - return $this; - } - - public function everySecond(): self - { - return $this->seconds(1); - } - - public function everyTwoSeconds(): self - { - return $this->seconds(2); - } - - public function everyFiveSeconds(): self - { - return $this->seconds(5); - } - - public function everyTenSeconds(): self - { - return $this->seconds(10); - } - - public function everyFifteenSeconds(): self - { - return $this->seconds(15); - } - - public function everyThirtySeconds(): self - { - return $this->seconds(30); - } - - public function everyMinute(): self - { - return $this->seconds(60); - } - - public function everyTwoMinutes(): self - { - return $this->seconds(120); - } - - public function everyFiveMinutes(): self - { - return $this->seconds(300); - } - - public function everyTenMinutes(): self - { - return $this->seconds(600); - } - - public function everyFifteenMinutes(): self - { - return $this->seconds(900); - } - - public function everyThirtyMinutes(): self - { - return $this->seconds(1800); - } - - public function hourly(): self - { - return $this->seconds(3600); - } - - public function reference(): self - { - $this->reference = true; - - if ($this->timer) { - $this->timer->reference(); - } - - return $this; - } - - public function unreference(): self - { - $this->reference = false; - - if ($this->timer) { - $this->timer->unreference(); - } - - return $this; - } - - public function run(): self - { - $this->timer = new Interval($this->interval, $this->closure, $this->reference); - - return $this; - } - - public function enable(): self - { - if ($this->timer) { - $this->timer->enable(); - } - - return $this; - } - - public function disable(): self - { - if ($this->timer) { - $this->timer->disable(); - } - - return $this; - } - - public function isEnabled(): bool - { - return $this->timer?->isEnabled() ?? false; - } -} diff --git a/src/Scheduling/TimerRegistry.php b/src/Scheduling/TimerRegistry.php deleted file mode 100644 index f8e6fc22..00000000 --- a/src/Scheduling/TimerRegistry.php +++ /dev/null @@ -1,27 +0,0 @@ - - */ - protected static array $timers = []; - - public static function add(Timer $timer): void - { - self::$timers[] = $timer; - } - - public static function run(): void - { - foreach (self::$timers as $timer) { - $timer->run(); - } - - self::$timers = []; - } -} diff --git a/tests/Feature/Queue/RedisQueueIntegrationTest.php b/tests/Feature/Queue/RedisQueueIntegrationTest.php new file mode 100644 index 00000000..3a4741d1 --- /dev/null +++ b/tests/Feature/Queue/RedisQueueIntegrationTest.php @@ -0,0 +1,174 @@ +execute('PING'); + } catch (Throwable $exception) { + $connectionError = $exception->getMessage(); + + throw new RuntimeException($connectionError, previous: $exception); + } + + return $client; +} + +beforeEach(function (): void { + try { + $this->redis = redisQueueIntegrationClient(); + } catch (Throwable $exception) { + $this->markTestSkipped('A real Redis server is required: ' . $exception->getMessage()); + } + + $this->queueName = 'integration-' . bin2hex(random_bytes(8)); + $this->taskIds = []; +}); + +afterEach(function (): void { + if (! isset($this->redis, $this->queueName)) { + return; + } + + $keys = [ + "queues:{$this->queueName}:ready", + "queues:{$this->queueName}:reserved", + "queues:{$this->queueName}:delayed", + ]; + + foreach ($this->taskIds as $taskId) { + $keys[] = "task:data:{$taskId}"; + $keys[] = "task:failed:{$taskId}"; + $this->redis->execute('LREM', 'queues:failed', 0, $taskId); + } + + $this->redis->execute('DEL', ...$keys); +}); + +it('allows only one worker to reserve a ready task', function (): void { + $task = new BasicQueuableTask(); + $this->taskIds[] = (string) $task->getTaskId(); + $firstQueue = new RedisQueue($this->redis, $this->queueName); + $secondQueue = new RedisQueue($this->redis, $this->queueName); + + $firstQueue->push($task); + + $reserved = $firstQueue->pop(); + + expect($reserved)->toBeInstanceOf(BasicQueuableTask::class) + ->and($secondQueue->pop())->toBeNull(); + + assert($reserved instanceof BasicQueuableTask); + $firstQueue->getStateManager()->complete($reserved); +}); + +it('recovers an expired reservation and rejects its stale acknowledgement', function (): void { + $task = new BasicQueuableTask(); + $taskId = (string) $task->getTaskId(); + $this->taskIds[] = $taskId; + $firstQueue = new RedisQueue($this->redis, $this->queueName, 1); + $secondQueue = new RedisQueue($this->redis, $this->queueName, 1); + + $firstQueue->push($task); + $staleReservation = $firstQueue->pop(); + + sleep(2); + + $activeReservation = $secondQueue->pop(); + assert($staleReservation instanceof BasicQueuableTask); + assert($activeReservation instanceof BasicQueuableTask); + + expect($activeReservation)->toBeInstanceOf(BasicQueuableTask::class) + ->and($activeReservation->getTaskId())->toBe($taskId) + ->and($activeReservation->getAttempts())->toBe(2); + + $firstQueue->getStateManager()->complete($staleReservation); + + expect($secondQueue->getStateManager()->getTaskState($taskId))->not->toBeNull(); + + $secondQueue->getStateManager()->complete($activeReservation); + + expect($secondQueue->getStateManager()->getTaskState($taskId))->toBeNull(); +}); + +it('clears pending tasks without deleting an active reservation', function (): void { + $active = new BasicQueuableTask(); + $pending = new BasicQueuableTask(); + $activeId = (string) $active->getTaskId(); + $pendingId = (string) $pending->getTaskId(); + $this->taskIds = [$activeId, $pendingId]; + $queue = new RedisQueue($this->redis, $this->queueName); + + $queue->push($active); + $queue->push($pending); + $reservation = $queue->pop(); + $queue->clear(); + + expect($queue->size())->toBe(0) + ->and($queue->getStateManager()->getTaskState($activeId))->not->toBeNull() + ->and($queue->getStateManager()->getTaskState($pendingId))->toBeNull(); + + assert($reservation instanceof BasicQueuableTask); + $queue->getStateManager()->complete($reservation); +}); + +it('promotes a delayed retry back to its original queue', function (): void { + $task = new BasicQueuableTask(); + $this->taskIds[] = (string) $task->getTaskId(); + $queue = new RedisQueue($this->redis, $this->queueName); + + $queue->push($task); + $reservation = $queue->pop(); + assert($reservation instanceof BasicQueuableTask); + $queue->getStateManager()->retry($reservation, 1); + + expect($queue->pop())->toBeNull(); + + sleep(2); + + $retried = $queue->pop(); + assert($retried instanceof BasicQueuableTask); + + expect($retried)->toBeInstanceOf(BasicQueuableTask::class) + ->and($retried->getAttempts())->toBe(2); + + $queue->getStateManager()->complete($retried); +}); + +it('moves a terminal failure to the failed queue atomically', function (): void { + $task = new BasicQueuableTask(); + $taskId = (string) $task->getTaskId(); + $this->taskIds[] = $taskId; + $queue = new RedisQueue($this->redis, $this->queueName); + + $queue->push($task); + $reservation = $queue->pop(); + assert($reservation instanceof BasicQueuableTask); + + $queue->getStateManager()->fail($reservation, new RuntimeException('Terminal failure')); + + expect($queue->getStateManager()->getTaskState($taskId))->toBeNull() + ->and($this->redis->execute('HGET', "task:failed:{$taskId}", 'exception'))->toContain('Terminal failure') + ->and($this->redis->execute('LRANGE', 'queues:failed', 0, -1))->toContain($taskId); +}); diff --git a/tests/Internal/FakeScheduleLock.php b/tests/Internal/FakeScheduleLock.php new file mode 100644 index 00000000..ed6cc38e --- /dev/null +++ b/tests/Internal/FakeScheduleLock.php @@ -0,0 +1,38 @@ + */ + protected array $locks = []; + + protected int $sequence = 0; + + public function acquire(string $schedule, string $occurrence, int $ttl): string|null + { + $key = "{$schedule}:{$occurrence}"; + + if (isset($this->locks[$key])) { + return null; + } + + $token = 'token-' . ++$this->sequence; + $this->locks[$key] = $token; + + return $token; + } + + public function release(string $schedule, string $occurrence, string $token): void + { + $key = "{$schedule}:{$occurrence}"; + + if (($this->locks[$key] ?? null) === $token) { + unset($this->locks[$key]); + } + } +} diff --git a/tests/Unit/Queue/RedisQueueTest.php b/tests/Unit/Queue/RedisQueueTest.php index 5bdf13fc..2771c8b1 100644 --- a/tests/Unit/Queue/RedisQueueTest.php +++ b/tests/Unit/Queue/RedisQueueTest.php @@ -2,479 +2,185 @@ declare(strict_types=1); -use Phenix\Database\Constants\Connection; -use Phenix\Facades\Config; -use Phenix\Facades\Queue; -use Phenix\Queue\Constants\QueueDriver; -use Phenix\Queue\QueueManager; +use Phenix\Queue\LuaScripts; use Phenix\Queue\RedisQueue; use Phenix\Queue\StateManagers\RedisTaskState; use Phenix\Redis\ClientWrapper; use Tests\Unit\Tasks\Internal\BasicQueuableTask; -beforeEach(function (): void { - Config::set('queue.default', QueueDriver::REDIS->value); -}); - -it('dispatch a task', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - $clientMock->expects($this->once()) - ->method('execute') - ->with( - $this->equalTo('RPUSH'), - $this->equalTo('queues:default'), - $this->isType('string') // Assuming payload is serialized to a string - ) - ->willReturn(true); - - $this->app->swap(Connection::redis('default'), $clientMock); - - BasicQueuableTask::dispatch(); -}); - -it('push the task', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - $clientMock->expects($this->once()) - ->method('execute') - ->with( - $this->equalTo('RPUSH'), - $this->equalTo('queues:default'), - $this->isType('string') - ) - ->willReturn(true); - - $this->app->swap(Connection::redis('default'), $clientMock); - - Queue::push(new BasicQueuableTask()); -}); - -it('enqueues the task on a custom queue', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); +it('stores task data and the ready id atomically', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $task = new BasicQueuableTask(); - $clientMock->expects($this->once()) + $client->expects($this->once()) ->method('execute') ->with( - $this->equalTo('RPUSH'), - $this->equalTo('queues:custom-queue'), - $this->isType('string') - ) - ->willReturn(true); - - $this->app->swap(Connection::redis('default'), $clientMock); - - Queue::pushOn('custom-queue', new BasicQueuableTask()); -}); - -it('returns a task', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - $payload = serialize(new BasicQueuableTask()); - - $clientMock->expects($this->exactly(4)) - ->method('execute') - ->withConsecutive( - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default'), $this->equalTo('queues:failed'), $this->equalTo('queues:delayed'), $this->isType('int'), $this->equalTo(60)], - [$this->equalTo('SETNX'), $this->stringStartsWith('task:reserved:'), $this->isType('int')], - [ - $this->equalTo('HSET'), - $this->stringStartsWith('task:data:'), - $this->isType('string'), // attempts - $this->isType('int'), // 1 - $this->isType('string'), // reserved_at - $this->isType('int'), // timestamp - $this->isType('string'), // reserved_until - $this->isType('int'), // timestamp - $this->isType('string'), // payload - $this->isType('string'), // serialized payload - ], - [$this->equalTo('EXPIRE'), $this->stringStartsWith('task:data:'), $this->isType('int')] - ) - ->willReturnOnConsecutiveCalls( - $payload, - 1, - 1, - 1 + 'EVAL', + $this->isType('string'), + 2, + 'queues:default:ready', + 'task:data:' . $task->getTaskId(), + $task->getTaskId(), + $this->isType('string'), + 'default', + $this->isType('int') ); - $this->app->swap(Connection::redis('default'), $clientMock); - - $task = Queue::pop(); - expect($task)->not()->toBeNull(); - expect($task)->toBeInstanceOf(BasicQueuableTask::class); + (new RedisQueue($client))->push($task); }); -it('rejects unauthorized serialized payloads', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); +it('atomically pops and reserves a task', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $task = new BasicQueuableTask(); - $clientMock->expects($this->once()) + $client->expects($this->once()) ->method('execute') ->with( - $this->equalTo('EVAL'), + 'EVAL', $this->isType('string'), - $this->equalTo(3), - $this->equalTo('queues:default'), - $this->equalTo('queues:failed'), - $this->equalTo('queues:delayed'), + 3, + 'queues:default:ready', + 'queues:default:reserved', + 'queues:default:delayed', + $this->isType('int'), $this->isType('int'), - $this->equalTo(60) + $this->isType('string') ) - ->willReturn(serialize(new stdClass())); - - $this->app->swap(Connection::redis('default'), $clientMock); - - expect(Queue::pop())->toBeNull(); -}); - -it('returns the queue size', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - $clientMock->expects($this->once()) - ->method('execute') - ->with($this->equalTo('LLEN'), $this->equalTo('queues:default')) - ->willReturn(7); - - $this->app->swap(Connection::redis('default'), $clientMock); - - expect(Queue::size())->toBe(7); -}); - -it('clear the queue', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); + ->willReturn([$task->getTaskId(), $task->getPayload(), 2]); - $clientMock->expects($this->once()) - ->method('execute') - ->with($this->equalTo('DEL'), $this->equalTo('queues:default')); - - $this->app->swap(Connection::redis('default'), $clientMock); + $popped = (new RedisQueue($client))->pop(); - Queue::clear(); + expect($popped)->toBeInstanceOf(BasicQueuableTask::class) + ->and($popped->getTaskId())->toBe($task->getTaskId()) + ->and($popped->getQueueName())->toBe('default') + ->and($popped->getAttempts())->toBe(2); }); -it('gets and sets the connection name via facade', function (): void { - $managerMock = $this->getMockBuilder(QueueManager::class) - ->disableOriginalConstructor() - ->getMock(); - - $managerMock->expects($this->once()) - ->method('getConnectionName') - ->willReturn('redis-connection'); - - $managerMock->expects($this->once()) - ->method('setConnectionName') - ->with('redis-connection'); - - $this->app->swap(QueueManager::class, $managerMock); - - expect(Queue::getConnectionName())->toBe('redis-connection'); +it('returns null when the ready queue is empty', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $client->expects($this->once())->method('execute')->willReturn(null); - Queue::setConnectionName('redis-connection'); + expect((new RedisQueue($client))->pop())->toBeNull(); }); -it('requeues the payload and returns null when reservation fails', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - $payload = serialize(new BasicQueuableTask()); - - $clientMock->expects($this->exactly(3)) - ->method('execute') - ->withConsecutive( - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default'), $this->equalTo('queues:failed'), $this->equalTo('queues:delayed'), $this->isType('int'), $this->equalTo(60)], - [$this->equalTo('SETNX'), $this->stringStartsWith('task:reserved:'), $this->isType('int')], - [$this->equalTo('RPUSH'), $this->equalTo('queues:default'), $this->identicalTo($payload)], - ) - ->willReturnOnConsecutiveCalls( - $payload, // EVAL returns a task payload (script handles failed task checking) - 0, // SETNX fails -> cannot reserve - 1 // RPUSH requeues the same payload - ); - - $this->app->swap(Connection::redis('default'), $clientMock); +it('reports ready queue size', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $client->expects($this->once())->method('execute') + ->with('LLEN', 'queues:default:ready')->willReturn(3); - $task = Queue::pop(); - - expect($task)->toBeNull(); + expect((new RedisQueue($client))->size())->toBe(3); }); -it('returns null when queue is empty', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - $clientMock->expects($this->once()) - ->method('execute') - ->with($this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default'), $this->equalTo('queues:failed'), $this->equalTo('queues:delayed'), $this->isType('int'), $this->equalTo(60)) - ->willReturn(null); // EVAL returns null when queue is empty or all tasks are failed - - $queue = new RedisQueue($clientMock, 'default'); - - $task = $queue->pop(); +it('clears ready and delayed tasks without deleting active reservations', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $client->expects($this->once())->method('execute')->with( + 'EVAL', + $this->isType('string'), + 3, + 'queues:default:ready', + 'queues:default:reserved', + 'queues:default:delayed' + ); - expect($task)->toBeNull(); + (new RedisQueue($client))->clear(); }); -it('marks a task as failed and cleans reservation/data keys', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - +it('acknowledges a reserved task atomically', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); $task = new BasicQueuableTask(); - $task->setTaskId('task-123'); + $task->setQueueName('emails'); + $state = new RedisTaskState($client); + $state->registerReceipt($task, 'receipt-1'); - $state = new RedisTaskState($clientMock); + $client->expects($this->once())->method('execute')->with( + 'EVAL', + $this->isType('string'), + 2, + 'queues:emails:reserved', + 'task:data:' . $task->getTaskId(), + $task->getTaskId(), + 'receipt-1' + ); - $clientMock->expects($this->exactly(3)) - ->method('execute') - ->withConsecutive( - [ - $this->equalTo('HSET'), - $this->equalTo('task:failed:task-123'), - $this->equalTo('task_id'), $this->equalTo('task-123'), - $this->equalTo('failed_at'), $this->isType('int'), - $this->equalTo('exception'), $this->isType('string'), - $this->equalTo('payload'), $this->isType('string'), - ], - [ - $this->equalTo('LPUSH'), - $this->equalTo('queues:failed'), - $this->equalTo('task-123'), - ], - [ - $this->equalTo('DEL'), - $this->equalTo('task:reserved:task-123'), - $this->equalTo('task:data:task-123'), - ] - ) - ->willReturnOnConsecutiveCalls(1, 1, 1); - - $state->fail($task, new Exception('Boom', 500)); + $state->complete($task); }); -it('retries a task with delay greater than zero by enqueuing into the delayed zset', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - +it('retries into the queue-specific delayed set', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); $task = new BasicQueuableTask(); - $task->setTaskId('task-retry-1'); + $task->setQueueName('emails'); + $task->setAttempts(2); + $state = new RedisTaskState($client); + $state->registerReceipt($task, 'receipt-2'); - $clientMock->expects($this->exactly(3)) - ->method('execute') - ->withConsecutive( - [ - $this->equalTo('EVAL'), - $this->isType('string'), // Lua script - $this->equalTo(4), // number of keys - $this->equalTo('task:reserved:task-retry-1'), - $this->equalTo('task:data:task-retry-1'), - $this->equalTo('queues:'), - $this->equalTo('queues:delayed'), - $this->equalTo(1), // attempts should be 1 after increment - $this->callback(function ($payload) { - // The payload should contain a task with attempts = 1 - $task = unserialize($payload); - - return $task->getAttempts() === 1; - }), - $this->equalTo(30), // delay - $this->isType('int'), // execute_at timestamp - ], - [ - $this->equalTo('DEL'), - $this->equalTo('task:failed:task-retry-1'), - ], - [ - $this->equalTo('LREM'), - $this->equalTo('queues:failed'), - $this->equalTo(0), - $this->equalTo('task-retry-1'), - ], - ) - ->willReturnOnConsecutiveCalls(1, 1, 1); + $client->expects($this->once())->method('execute')->with( + 'EVAL', + $this->isType('string'), + 6, + 'queues:emails:reserved', + 'task:data:' . $task->getTaskId(), + 'queues:emails:ready', + 'queues:emails:delayed', + 'task:failed:' . $task->getTaskId(), + 'queues:failed', + $task->getTaskId(), + $this->callback(fn (string $payload): bool => unserialize($payload)->getAttempts() === 2), + 30, + $this->isType('int'), + 'receipt-2' + ); - $queue = new RedisQueue($clientMock); - $queue->getStateManager()->retry($task, 30); + $state->retry($task, 30); }); -it('cleans expired reservations via Lua script', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - $clientMock->expects($this->once()) - ->method('execute') - ->with( - $this->equalTo('EVAL'), - $this->isType('string'), // Lua script - $this->equalTo(0), - $this->isType('int'), // now timestamp - ) - ->willReturn(1); +it('recovers expired reservations in bounded batches during pop', function (): void { + $script = LuaScripts::pop(); - $state = new RedisTaskState($clientMock); - $state->cleanupExpiredReservations(); + expect($script)->toContain("redis.call('ZRANGEBYSCORE', KEYS[2], 0, now, 'LIMIT', 0, 100)") + ->and($script)->toContain("redis.call('RPUSH', KEYS[1], task_id)") + ->and($script)->not->toContain("redis.call('SCAN'"); }); -it('returns null from getTaskState when no data exists', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - $clientMock->expects($this->once()) - ->method('execute') - ->with($this->equalTo('HGETALL'), $this->equalTo('task:data:task-nope')) - ->willReturn([]); - - $state = new RedisTaskState($clientMock); - $this->assertNull($state->getTaskState('task-nope')); +it('rejects stale acknowledgements using the reservation receipt', function (): void { + expect(LuaScripts::complete())->toContain("'receipt') ~= ARGV[2]") + ->and(LuaScripts::retry())->toContain("'receipt') ~= ARGV[5]") + ->and(LuaScripts::fail())->toContain("'receipt') ~= ARGV[6]"); }); -it('returns task state array from getTaskState when data exists', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - // Simulate Redis HGETALL flat array response - $hgetAll = [ - 'attempts', 2, - 'reserved_at', 1700000000, - 'available_at', 1700000100, - 'payload', serialize(new BasicQueuableTask()), - ]; +it('keeps queue ownership while promoting delayed tasks', function (): void { + $script = LuaScripts::pop(); - $clientMock->expects($this->once()) - ->method('execute') - ->with($this->equalTo('HGETALL'), $this->equalTo('task:data:task-123')) - ->willReturn($hgetAll); - - $state = new RedisTaskState($clientMock); - $data = $state->getTaskState('task-123'); - - $this->assertIsArray($data); - $this->assertArrayHasKey('attempts', $data); - $this->assertArrayHasKey('reserved_at', $data); - $this->assertArrayHasKey('available_at', $data); - $this->assertArrayHasKey('payload', $data); - $this->assertSame(2, $data['attempts']); - $this->assertSame(1700000000, $data['reserved_at']); - $this->assertSame(1700000100, $data['available_at']); - $this->assertIsString($data['payload']); + expect($script)->toContain("redis.call('ZRANGEBYSCORE', KEYS[3]") + ->and($script)->toContain("redis.call('RPUSH', KEYS[1], task_id)"); }); -it('properly pops tasks in chunks with limited timeout', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); +it('associates receipts with reservation objects instead of task ids', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $first = new BasicQueuableTask(); + $second = clone $first; + $state = new RedisTaskState($client); - $queue = new RedisQueue($clientMock, 'default'); + $state->registerReceipt($first, 'receipt-1'); + $state->registerReceipt($second, 'receipt-2'); - $payloads = [ - serialize(new BasicQueuableTask()), - serialize(new BasicQueuableTask()), - serialize(new BasicQueuableTask()), - ]; + $client->expects($this->exactly(2))->method('execute')->withConsecutive( + ['EVAL', $this->isType('string'), 2, 'queues:default:reserved', 'task:data:' . $first->getTaskId(), $first->getTaskId(), 'receipt-1'], + ['EVAL', $this->isType('string'), 2, 'queues:default:reserved', 'task:data:' . $second->getTaskId(), $second->getTaskId(), 'receipt-2'] + ); - // Expect 12 calls to execute method with specific order - $clientMock->expects($this->exactly(12)) - ->method('execute') - ->withConsecutive( - // First task - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default'), $this->equalTo('queues:failed'), $this->equalTo('queues:delayed'), $this->isType('int'), $this->equalTo(60)], - [$this->equalTo('SETNX'), $this->stringStartsWith('task:reserved:'), $this->isType('int')], - [$this->equalTo('HSET'), $this->stringStartsWith('task:data:'), $this->isType('string'), $this->isType('int'), $this->isType('string'), $this->isType('int'), $this->isType('string'), $this->isType('int'), $this->isType('string'), $this->isType('string')], - [$this->equalTo('EXPIRE'), $this->stringStartsWith('task:data:'), $this->isType('int')], - // Second task - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default'), $this->equalTo('queues:failed'), $this->equalTo('queues:delayed'), $this->isType('int'), $this->equalTo(60)], - [$this->equalTo('SETNX'), $this->stringStartsWith('task:reserved:'), $this->isType('int')], - [$this->equalTo('HSET'), $this->stringStartsWith('task:data:'), $this->isType('string'), $this->isType('int'), $this->isType('string'), $this->isType('int'), $this->isType('string'), $this->isType('int'), $this->isType('string'), $this->isType('string')], - [$this->equalTo('EXPIRE'), $this->stringStartsWith('task:data:'), $this->isType('int')], - // Third task - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default'), $this->equalTo('queues:failed'), $this->equalTo('queues:delayed'), $this->isType('int'), $this->equalTo(60)], - [$this->equalTo('SETNX'), $this->stringStartsWith('task:reserved:'), $this->isType('int')], - [$this->equalTo('HSET'), $this->stringStartsWith('task:data:'), $this->isType('string'), $this->isType('int'), $this->isType('string'), $this->isType('int'), $this->isType('string'), $this->isType('int'), $this->isType('string'), $this->isType('string')], - [$this->equalTo('EXPIRE'), $this->stringStartsWith('task:data:'), $this->isType('int')] - ) - ->willReturnOnConsecutiveCalls( - // First task returns - $payloads[0], - 1, - 1, - 1, - // Second task returns - $payloads[1], - 1, - 1, - 1, - // Third task returns - $payloads[2], - 1, - 1, - 1 - ); - - $task1 = $queue->pop(); - $task2 = $queue->pop(); - $task3 = $queue->pop(); - - expect($task1)->toBeInstanceOf(BasicQueuableTask::class); - expect($task2)->toBeInstanceOf(BasicQueuableTask::class); - expect($task3)->toBeInstanceOf(BasicQueuableTask::class); + $state->complete($first); + $state->complete($second); }); -it('returns empty chunk when limit is zero', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); +it('does not report an unregistered task as reserved', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); - $clientMock->expects($this->never())->method('execute'); - - $queue = new RedisQueue($clientMock); - $chunk = $queue->popChunk(0); - - $this->assertIsArray($chunk); - $this->assertCount(0, $chunk); + expect((new RedisTaskState($client))->reserve(new BasicQueuableTask()))->toBeFalse(); }); -it('returns empty chunk when first reservation fails', function (): void { - $clientMock = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); - - $payload1 = serialize(new BasicQueuableTask()); // Will fail reservation - - $clientMock->expects($this->exactly(3)) - ->method('execute') - ->withConsecutive( - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default'), $this->equalTo('queues:failed'), $this->equalTo('queues:delayed'), $this->isType('int'), $this->equalTo(60)], - [$this->equalTo('SETNX'), $this->stringStartsWith('task:reserved:'), $this->isType('int')], - [$this->equalTo('RPUSH'), $this->equalTo('queues:default'), $this->identicalTo($payload1)], - ) - ->willReturnOnConsecutiveCalls( - $payload1, // EVAL returns payload (script handles failed task checking) - 0, // SETNX fails - 1, // RPUSH requeues payload - ); - - $queue = new RedisQueue($clientMock); - $chunk = $queue->popChunk(3); +it('returns empty chunk for a non-positive limit', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $client->expects($this->never())->method('execute'); - $this->assertIsArray($chunk); - $this->assertCount(0, $chunk); + expect((new RedisQueue($client))->popChunk(0))->toBe([]); }); diff --git a/tests/Unit/Queue/WorkerRedisTest.php b/tests/Unit/Queue/WorkerRedisTest.php index 989c323a..f4523729 100644 --- a/tests/Unit/Queue/WorkerRedisTest.php +++ b/tests/Unit/Queue/WorkerRedisTest.php @@ -12,109 +12,36 @@ use Tests\Unit\Tasks\Internal\BadTask; use Tests\Unit\Tasks\Internal\BasicQueuableTask; -beforeEach(function () { +beforeEach(function (): void { Config::set('queue.default', QueueDriver::REDIS->value); }); -it('processes a successful task', function (): void { - $client = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); +it('processes and acknowledges a successful reserved task', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $task = new BasicQueuableTask(); - $payload = serialize(new BasicQueuableTask()); - - $client->expects($this->exactly(6)) - ->method('execute') + $client->expects($this->exactly(2))->method('execute') ->withConsecutive( - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default'), $this->equalTo('queues:failed'), $this->equalTo('queues:delayed'), $this->isType('int'), $this->equalTo(60)], - [$this->equalTo('SETNX'), $this->stringStartsWith('task:reserved:'), $this->isType('int')], - [ - $this->equalTo('HSET'), - $this->stringStartsWith('task:data:'), - $this->isType('string'), // attempts - $this->isType('int'), // 1 - $this->isType('string'), // reserved_at - $this->isType('int'), // timestamp - $this->isType('string'), // reserved_until - $this->isType('int'), // timestamp - $this->isType('string'), // payload - $this->isType('string'), // serialized payload - ], - [$this->equalTo('EXPIRE'), $this->stringStartsWith('task:data:'), $this->isType('int')], - [$this->equalTo('DEL'), $this->stringStartsWith('task:reserved:'), $this->stringStartsWith('task:data:')], - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(0), $this->isType('int')] + [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default:ready'), $this->equalTo('queues:default:reserved'), $this->equalTo('queues:default:delayed'), $this->isType('int'), $this->isType('int'), $this->isType('string')], + [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(2), $this->equalTo('queues:default:reserved'), $this->equalTo('task:data:' . $task->getTaskId()), $this->equalTo($task->getTaskId()), $this->isType('string')] ) - ->willReturnOnConsecutiveCalls( - $payload, // EVAL returns payload (script handles failed task checking) - 1, // SETNX succeeds - 1, // HSET succeeds - 1, // EXPIRE succeeds - 1, // DEL succeeds - 1 // EVAL cleanup succeeds - ); + ->willReturnOnConsecutiveCalls([$task->getTaskId(), $task->getPayload(), 1], 1); $this->app->swap(Connection::redis('default'), $client); - - $queueManager = new QueueManager(); - $worker = new Worker($queueManager); - - $worker->runOnce('default', 'default', new WorkerOptions(once: true, sleep: 1)); + (new Worker(new QueueManager()))->runOnce('default', 'default', new WorkerOptions(once: true)); }); -it('processes a failed task and retries', function (): void { - $client = $this->getMockBuilder(ClientWrapper::class) - ->disableOriginalConstructor() - ->getMock(); +it('atomically schedules retry for a failed reserved task', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $task = new BadTask(); - $payload = serialize(new BadTask()); - - $client->expects($this->exactly(8)) - ->method('execute') + $client->expects($this->exactly(2))->method('execute') ->withConsecutive( - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default'), $this->equalTo('queues:failed'), $this->equalTo('queues:delayed'), $this->isType('int'), $this->equalTo(60)], - [$this->equalTo('SETNX'), $this->stringStartsWith('task:reserved:'), $this->isType('int')], - [ - $this->equalTo('HSET'), - $this->stringStartsWith('task:data:'), - $this->isType('string'), $this->isType('int'), - $this->isType('string'), $this->isType('int'), - $this->isType('string'), $this->isType('int'), - $this->isType('string'), $this->isType('string'), - ], - [$this->equalTo('EXPIRE'), $this->stringStartsWith('task:data:'), $this->isType('int')], - // retry() - now uses Lua script - [ - $this->equalTo('EVAL'), - $this->isType('string'), // Lua script - $this->equalTo(4), // number of keys - $this->stringStartsWith('task:reserved:'), - $this->stringStartsWith('task:data:'), - $this->equalTo('queues:default'), - $this->equalTo('queues:delayed'), - $this->isType('int'), // attempts - $this->isType('string'), // payload - $this->equalTo(0), // delay - $this->isType('int'), // execute_at timestamp - ], - [$this->equalTo('DEL'), $this->stringStartsWith('task:failed:')], - [$this->equalTo('LREM'), $this->equalTo('queues:failed'), $this->equalTo(0), $this->isType('string')], - [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(0), $this->isType('int')], // cleanupExpiredReservations at end of processTask + [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(3), $this->equalTo('queues:default:ready'), $this->equalTo('queues:default:reserved'), $this->equalTo('queues:default:delayed'), $this->isType('int'), $this->isType('int'), $this->isType('string')], + [$this->equalTo('EVAL'), $this->isType('string'), $this->equalTo(6), $this->equalTo('queues:default:reserved'), $this->equalTo('task:data:' . $task->getTaskId()), $this->equalTo('queues:default:ready'), $this->equalTo('queues:default:delayed'), $this->equalTo('task:failed:' . $task->getTaskId()), $this->equalTo('queues:failed'), $this->equalTo($task->getTaskId()), $this->isType('string'), $this->equalTo(0), $this->isType('int'), $this->isType('string')] ) - ->willReturnOnConsecutiveCalls( - $payload, // EVAL returns payload (script handles failed task checking) - 1, // SETNX succeeds - 1, // HSET succeeds - 1, // EXPIRE succeeds - 1, // EVAL succeeds (retry Lua script) - 1, // DEL succeeds (cleanup failed) - 1, // LREM succeeds (cleanup failed queue) - 1 // EVAL cleanup succeeds - ); + ->willReturnOnConsecutiveCalls([$task->getTaskId(), $task->getPayload(), 1], 1); $this->app->swap(Connection::redis('default'), $client); - - $queueManager = new QueueManager(); - $worker = new Worker($queueManager); - - $worker->runOnce('default', 'default', new WorkerOptions(once: true, sleep: 1, retryDelay: 0)); + (new Worker(new QueueManager()))->runOnce('default', 'default', new WorkerOptions(once: true, retryDelay: 0)); }); diff --git a/tests/Unit/Scheduling/Console/ScheduleRunCommandTest.php b/tests/Unit/Scheduling/Console/ScheduleRunCommandTest.php index a207fbca..cc979d18 100644 --- a/tests/Unit/Scheduling/Console/ScheduleRunCommandTest.php +++ b/tests/Unit/Scheduling/Console/ScheduleRunCommandTest.php @@ -3,12 +3,16 @@ declare(strict_types=1); use Phenix\Facades\Schedule; +use Phenix\Scheduling\Contracts\ScheduleLock; use Symfony\Component\Console\Tester\CommandTester; +use Tests\Internal\FakeScheduleLock; it('run schedule once', function (): void { $executed = false; - Schedule::call(function () use (&$executed): void { + $this->app->swap(ScheduleLock::class, new FakeScheduleLock()); + + Schedule::call('schedule-run-command-test', function () use (&$executed): void { $executed = true; })->everyMinute(); diff --git a/tests/Unit/Scheduling/RedisScheduleLockTest.php b/tests/Unit/Scheduling/RedisScheduleLockTest.php new file mode 100644 index 00000000..91b403c9 --- /dev/null +++ b/tests/Unit/Scheduling/RedisScheduleLockTest.php @@ -0,0 +1,46 @@ +getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + + $client->expects($this->once())->method('execute')->with( + 'EVAL', + $this->stringContains("redis.call('SET'"), + 1, + 'test:schedule:' . hash('sha256', 'reconcile-usage') . ':202610061430', + 86400, + $this->isType('string') + )->willReturn(1); + + $lock = new RedisScheduleLock($client, 'test:schedule:'); + + expect($lock->acquire('reconcile-usage', '202610061430', 86400))->toBeString(); +}); + +it('reports an occurrence already acquired by another scheduler', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $client->expects($this->once())->method('execute')->willReturn(0); + + $lock = new RedisScheduleLock($client); + + expect($lock->acquire('reconcile-usage', '202610061430', 86400))->toBeNull(); +}); + +it('releases only the lock owned by the supplied token', function (): void { + $client = $this->getMockBuilder(ClientWrapper::class)->disableOriginalConstructor()->getMock(); + $client->expects($this->once())->method('execute')->with( + 'EVAL', + $this->stringContains("redis.call('GET'"), + 1, + 'test:schedule:' . hash('sha256', 'reconcile-usage') . ':202610061430', + 'owner-token' + )->willReturn(1); + + (new RedisScheduleLock($client, 'test:schedule:')) + ->release('reconcile-usage', '202610061430', 'owner-token'); +}); diff --git a/tests/Unit/Scheduling/SchedulerTest.php b/tests/Unit/Scheduling/SchedulerTest.php index 70bce5ba..3706724e 100644 --- a/tests/Unit/Scheduling/SchedulerTest.php +++ b/tests/Unit/Scheduling/SchedulerTest.php @@ -5,13 +5,14 @@ use Phenix\Scheduling\Schedule; use Phenix\Scheduling\Scheduler; use Phenix\Util\Date; +use Tests\Internal\FakeScheduleLock; it('executes when expression is due (every minute)', function (): void { - $schedule = new Schedule(); + $schedule = new Schedule(new FakeScheduleLock()); $executed = false; - $scheduler = $schedule->call(function () use (&$executed): void { + $scheduler = $schedule->call('schedule-' . __LINE__, function () use (&$executed): void { $executed = true; })->everyMinute(); @@ -22,12 +23,58 @@ expect($executed)->toBeTrue(); }); +it('executes a named occurrence only once across scheduler instances', function (): void { + $lock = new FakeScheduleLock(); + $executions = 0; + $now = Date::now('UTC')->startOfMinute(); + + $first = (new Schedule($lock))->call('reconcile-usage', function () use (&$executions): void { + $executions++; + })->everyMinute(); + $second = (new Schedule($lock))->call('reconcile-usage', function () use (&$executions): void { + $executions++; + })->everyMinute(); + + $first->tick($now); + $second->tick($now); + + expect($executions)->toBe(1); +}); + +it('rejects duplicate schedule names in one application', function (): void { + $schedule = new Schedule(new FakeScheduleLock()); + $schedule->call('reconcile-usage', function (): void { + }); + + expect(fn () => $schedule->call('reconcile-usage', function (): void { + }))->toThrow(\InvalidArgumentException::class); +}); + +it('releases the occurrence lock when the callback throws', function (): void { + $lock = new FakeScheduleLock(); + $now = Date::now('UTC')->startOfMinute(); + $attempts = 0; + $scheduler = (new Schedule($lock))->call('retry-failed-dispatch', function () use (&$attempts): void { + $attempts++; + + if ($attempts === 1) { + throw new RuntimeException('Dispatch failed'); + } + })->everyMinute(); + + expect(fn () => $scheduler->tick($now))->toThrow(RuntimeException::class, 'Dispatch failed'); + + $scheduler->tick($now); + + expect($attempts)->toBe(2); +}); + it('does not execute when not due (dailyAt time mismatch)', function (): void { - $schedule = new Schedule(); + $schedule = new Schedule(new FakeScheduleLock()); $executed = false; - $scheduler = $schedule->call(function () use (&$executed): void { + $scheduler = $schedule->call('schedule-' . __LINE__, function () use (&$executed): void { $executed = true; })->dailyAt('10:15'); @@ -45,11 +92,11 @@ }); it('executes exactly at matching dailyAt time', function (): void { - $schedule = new Schedule(); + $schedule = new Schedule(new FakeScheduleLock()); $executed = false; - $scheduler = $schedule->call(function () use (&$executed): void { + $scheduler = $schedule->call('schedule-' . __LINE__, function () use (&$executed): void { $executed = true; })->dailyAt('10:15'); @@ -61,11 +108,11 @@ }); it('respects timezone when evaluating due', function (): void { - $schedule = new Schedule(); + $schedule = new Schedule(new FakeScheduleLock()); $executed = false; - $scheduler = $schedule->call(function () use (&$executed): void { + $scheduler = $schedule->call('schedule-' . __LINE__, function () use (&$executed): void { $executed = true; })->dailyAt('12:00')->timezone('America/New_York'); @@ -79,11 +126,11 @@ }); it('supports */5 minutes schedule and only runs on multiples of five', function (): void { - $schedule = new Schedule(); + $schedule = new Schedule(new FakeScheduleLock()); $executed = false; - $scheduler = $schedule->call(function () use (&$executed): void { + $scheduler = $schedule->call('schedule-' . __LINE__, function () use (&$executed): void { $executed = true; })->everyFiveMinutes(); @@ -103,9 +150,9 @@ it('does nothing when no expression is set', function (): void { $executed = false; - $scheduler = new Scheduler(function () use (&$executed): void { + $scheduler = new Scheduler('no-expression', function () use (&$executed): void { $executed = true; - }); + }, new FakeScheduleLock()); $now = Date::now('UTC')->startOfDay(); @@ -115,7 +162,7 @@ }); it('sets cron for weekly', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->weekly(); $ref = new ReflectionClass($scheduler); @@ -127,7 +174,7 @@ }); it('sets cron for monthly', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->monthly(); $ref = new ReflectionClass($scheduler); @@ -139,7 +186,7 @@ }); it('sets cron for every ten minutes', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->everyTenMinutes(); $ref = new ReflectionClass($scheduler); @@ -151,7 +198,7 @@ }); it('sets cron for every fifteen minutes', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->everyFifteenMinutes(); $ref = new ReflectionClass($scheduler); @@ -163,7 +210,7 @@ }); it('sets cron for every thirty minutes', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->everyThirtyMinutes(); $ref = new ReflectionClass($scheduler); @@ -175,7 +222,7 @@ }); it('sets cron for every two hours', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->everyTwoHours(); $ref = new ReflectionClass($scheduler); @@ -187,7 +234,7 @@ }); it('sets cron for every two days', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->everyTwoDays(); $ref = new ReflectionClass($scheduler); @@ -199,7 +246,7 @@ }); it('sets cron for every weekday', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->everyWeekday(); $ref = new ReflectionClass($scheduler); @@ -211,7 +258,7 @@ }); it('sets cron for every weekend', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->everyWeekend(); $ref = new ReflectionClass($scheduler); @@ -223,7 +270,7 @@ }); it('sets cron for mondays', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->mondays(); $ref = new ReflectionClass($scheduler); @@ -235,7 +282,7 @@ }); it('sets cron for fridays', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->fridays(); $ref = new ReflectionClass($scheduler); @@ -247,7 +294,7 @@ }); it('sets cron for weeklyAt at specific time', function (): void { - $scheduler = (new Schedule())->call(function (): void { + $scheduler = (new Schedule(new FakeScheduleLock()))->call('schedule-' . __LINE__, function (): void { })->weeklyAt('10:15'); $ref = new ReflectionClass($scheduler); diff --git a/tests/Unit/Scheduling/TimerTest.php b/tests/Unit/Scheduling/TimerTest.php deleted file mode 100644 index bce54059..00000000 --- a/tests/Unit/Scheduling/TimerTest.php +++ /dev/null @@ -1,288 +0,0 @@ -timer(function () use (&$count): void { - $count++; - })->everySecond(); - - $timer->reference(); - - TimerRegistry::run(); - - delay(2.2); - - expect($count)->toBeGreaterThanOrEqual(2); - - $timer->disable(); - - $afterDisable = $count; - - delay(1.5); - - expect($count)->toBe($afterDisable); -}); - -it('can be re-enabled after disable', function (): void { - $schedule = new Schedule(); - - $count = 0; - - $timer = $schedule->timer(function () use (&$count): void { - $count++; - })->everySecond(); - - TimerRegistry::run(); - - delay(1.1); - - expect($count)->toBeGreaterThanOrEqual(1); - - $timer->disable(); - - $paused = $count; - - delay(1.2); - - expect($count)->toBe($paused); - - $timer->enable(); - - delay(1.2); - - expect($count)->toBeGreaterThan($paused); - - $timer->disable(); -}); - -it('supports millisecond intervals', function (): void { - $schedule = new Schedule(); - - $count = 0; - - $timer = $schedule->timer(function () use (&$count): void { - $count++; - })->milliseconds(100); - - TimerRegistry::run(); - - delay(0.35); - - expect($count)->toBeGreaterThanOrEqual(2); - - $timer->disable(); -}); - -it('unreference does not prevent execution', function (): void { - $schedule = new Schedule(); - - $executed = false; - - $timer = $schedule->timer(function () use (&$executed): void { - $executed = true; - })->everySecond()->unreference(); - - TimerRegistry::run(); - - delay(1.2); - - expect($executed)->toBeTrue(); - - $timer->disable(); -}); - -it('reports enabled state correctly', function (): void { - $schedule = new Schedule(); - - $timer = $schedule->timer(function (): void { - // no-op - })->everySecond(); - - expect($timer->isEnabled())->toBeFalse(); - - TimerRegistry::run(); - - expect($timer->isEnabled())->toBeTrue(); - - $timer->disable(); - - expect($timer->isEnabled())->toBeFalse(); - - $timer->enable(); - - expect($timer->isEnabled())->toBeTrue(); - - $timer->disable(); -}); - -it('runs at given using facade', function (): void { - $timerExecuted = false; - - $timer = ScheduleFacade::timer(function () use (&$timerExecuted): void { - $timerExecuted = true; - })->everySecond(); - - TimerRegistry::run(); - - delay(2); - - expect($timerExecuted)->toBeTrue(); - - $timer->disable(); -}); - -it('sets interval for every two seconds', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyTwoSeconds(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(2.0); -}); - -it('sets interval for every five seconds', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyFiveSeconds(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(5.0); -}); - -it('sets interval for every ten seconds', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyTenSeconds(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(10.0); -}); - -it('sets interval for every fifteen seconds', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyFifteenSeconds(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(15.0); -}); - -it('sets interval for every thirty seconds', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyThirtySeconds(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(30.0); -}); - -it('sets interval for every minute', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyMinute(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(60.0); -}); - -it('sets interval for every two minutes', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyTwoMinutes(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(120.0); -}); - -it('sets interval for every five minutes', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyFiveMinutes(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(300.0); -}); - -it('sets interval for every ten minutes', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyTenMinutes(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(600.0); -}); - -it('sets interval for every fifteen minutes', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyFifteenMinutes(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(900.0); -}); - -it('sets interval for every thirty minutes', function (): void { - $timer = new Timer(function (): void { - }); - $timer->everyThirtyMinutes(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(1800.0); -}); - -it('sets interval for hourly', function (): void { - $timer = new Timer(function (): void { - }); - $timer->hourly(); - - $ref = new ReflectionClass($timer); - $prop = $ref->getProperty('interval'); - $prop->setAccessible(true); - - expect($prop->getValue($timer))->toBe(3600.0); -}); diff --git a/tests/fixtures/application/config/queue.php b/tests/fixtures/application/config/queue.php index 25ab32b6..390d3c31 100644 --- a/tests/fixtures/application/config/queue.php +++ b/tests/fixtures/application/config/queue.php @@ -24,6 +24,7 @@ 'redis' => [ 'connection' => env('REDIS_QUEUE_CONNECTION', static fn (): string => 'default'), 'queue' => env('REDIS_QUEUE', static fn (): string => 'default'), + 'reservation_timeout' => env('REDIS_QUEUE_RESERVATION_TIMEOUT', static fn (): int => 60), ], ], ]; diff --git a/tests/fixtures/application/config/schedule.php b/tests/fixtures/application/config/schedule.php new file mode 100644 index 00000000..08677ae3 --- /dev/null +++ b/tests/fixtures/application/config/schedule.php @@ -0,0 +1,9 @@ + 'default', + 'lock_prefix' => 'phenix:test:schedule:', + 'occurrence_ttl' => 86400, +];