Improve redis implementation, improve scheduler

.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

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

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();

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

src/Queue/LuaScripts.php

@@ -4,207 +4,142 @@
 
 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;
     }
 
     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;
     }
 }

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)
         );
     }
 }