From c0df224ffaede069439fea186ed5d15c7a984054 Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:38:50 +0000 Subject: [PATCH 01/24] fix: preserve canonical Redis service resolution Keep the framework canonical string keys for Redis and database services while narrowing each resolved value to its real runtime contract. This preserves container conventions and lets static analysis follow connector and lifecycle calls without changing bindings or resolution behavior. Document the same rule in AGENTS.md so future PHPStan fixes use local truthful narrowing instead of changing service keys or introducing global ignores. --- AGENTS.md | 1 + src/queue/src/QueueServiceProvider.php | 20 +++++++++++++------ .../RedisConnectionLifecycleListener.php | 3 +++ src/redis/src/RedisServiceProvider.php | 8 +++++++- src/reverb/src/ReverbServiceProvider.php | 6 +++++- 5 files changed, 30 insertions(+), 8 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index b7e41b351..cda10e43e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -682,6 +682,7 @@ Run `./vendor/bin/phpstan` and `./vendor/bin/php-cs-fixer fix` without flags — 6. **Wrong docblock types should be fixed**, not suppressed. Check the actual runtime behavior (extension docs, reflection, tests) to determine the correct type. 7. **Type decisions must be evidence-based.** See Development Conventions — check Laravel/Hyperf signatures and docblocks, then trace real control flow. Don't guess. 8. **Narrowing / suppression order.** When the code is correct but PHPStan can't follow it, in order: (1) fix the type signature or docblock; (2) `@var` to narrow to the correct runtime type; (3) a line- or identifier-scoped `@phpstan-ignore` (e.g. magic `__call`/`__get` forwarding). Never use `assert()` to narrow types, and never add a neon-wide rule on your own (see #9). + When a container string key is also a PHP class name, keep the canonical service key and use `@var` for its actual runtime type; do not change service resolution solely for PHPStan. 9. **Don't add patterns to `phpstan.neon.dist` on your own.** The neon file's global ignores cover fundamental framework patterns (Eloquent magic, generics, `new static`). Fix new phpstan errors at the source, not by masking them with new neon rules. Under rare circumstances a global suppression genuinely is the best choice — if you think one may be needed, STOP, explain why the error can't be fixed at the source or narrowed locally, and ask for approval before adding it. ## Porting Packages diff --git a/src/queue/src/QueueServiceProvider.php b/src/queue/src/QueueServiceProvider.php index 864cb19b4..ad746d3da 100644 --- a/src/queue/src/QueueServiceProvider.php +++ b/src/queue/src/QueueServiceProvider.php @@ -7,6 +7,8 @@ use Hypervel\Contracts\Database\ModelIdentifier; use Hypervel\Contracts\Debug\ExceptionHandler; use Hypervel\Contracts\Events\Dispatcher as EventDispatcher; +use Hypervel\Contracts\Redis\Factory as RedisFactory; +use Hypervel\Database\ConnectionResolverInterface; use Hypervel\Queue\Connectors\BackgroundConnector; use Hypervel\Queue\Connectors\BeanstalkdConnector; use Hypervel\Queue\Connectors\DatabaseConnector; @@ -229,9 +231,12 @@ protected function registerFailoverConnector(QueueManager $manager): void */ protected function registerDatabaseConnector(QueueManager $manager): void { - $manager->addConnector('database', fn () => new DatabaseConnector( - $this->app['db'], - )); + $manager->addConnector('database', function (): DatabaseConnector { + /** @var ConnectionResolverInterface $connections */ + $connections = $this->app->make('db'); + + return new DatabaseConnector($connections); + }); } /** @@ -239,9 +244,12 @@ protected function registerDatabaseConnector(QueueManager $manager): void */ protected function registerRedisConnector(QueueManager $manager): void { - $manager->addConnector('redis', fn () => new RedisConnector( - $this->app['redis'], - )); + $manager->addConnector('redis', function (): RedisConnector { + /** @var RedisFactory $redis */ + $redis = $this->app->make('redis'); + + return new RedisConnector($redis); + }); } /** diff --git a/src/redis/src/Listeners/RedisConnectionLifecycleListener.php b/src/redis/src/Listeners/RedisConnectionLifecycleListener.php index 914826dd1..6c24c9ff4 100644 --- a/src/redis/src/Listeners/RedisConnectionLifecycleListener.php +++ b/src/redis/src/Listeners/RedisConnectionLifecycleListener.php @@ -5,6 +5,7 @@ namespace Hypervel\Redis\Listeners; use Hypervel\Contracts\Container\Container as ContainerContract; +use Hypervel\Contracts\Redis\Factory as RedisFactory; use Hypervel\Redis\Pool\PoolFactory; use Hypervel\Redis\RedisManager; use Throwable; @@ -25,6 +26,7 @@ public function releaseTaskConnections(): void return; } + /** @var RedisFactory $manager */ $manager = $this->container->make('redis'); if ($manager instanceof RedisManager) { @@ -41,6 +43,7 @@ public function discardProcessConnections(): void if ($this->container->resolved('redis')) { try { + /** @var RedisFactory $manager */ $manager = $this->container->make('redis'); if ($manager instanceof RedisManager) { diff --git a/src/redis/src/RedisServiceProvider.php b/src/redis/src/RedisServiceProvider.php index 13a835152..09d0c87d2 100644 --- a/src/redis/src/RedisServiceProvider.php +++ b/src/redis/src/RedisServiceProvider.php @@ -4,6 +4,7 @@ namespace Hypervel\Redis; +use Hypervel\Contracts\Redis\Factory as RedisFactory; use Hypervel\Core\Events\BeforeServerFork; use Hypervel\Core\Events\BeforeWorkerStart; use Hypervel\Core\Events\TaskTerminated; @@ -26,7 +27,12 @@ public function register(): void $app->make(RedisSentinelFactory::class), )); - $this->app->bind('redis.connection', fn ($app) => $app->make('redis')->connection()); + $this->app->bind('redis.connection', function ($app) { + /** @var RedisFactory $redis */ + $redis = $app->make('redis'); + + return $redis->connection(); + }); } /** diff --git a/src/reverb/src/ReverbServiceProvider.php b/src/reverb/src/ReverbServiceProvider.php index 57b1edbbb..529254fe0 100644 --- a/src/reverb/src/ReverbServiceProvider.php +++ b/src/reverb/src/ReverbServiceProvider.php @@ -5,6 +5,7 @@ namespace Hypervel\Reverb; use Hypervel\Contracts\Bus\Dispatcher as BusDispatcher; +use Hypervel\Contracts\Redis\Factory as RedisFactory; use Hypervel\Coordinator\Timer; use Hypervel\Core\Events\AfterWorkerStart; use Hypervel\Core\Events\OnPipeMessage; @@ -89,8 +90,11 @@ public function register(): void $connectionName = $app->make('config') ->string('reverb.servers.reverb.scaling.connection', 'reverb'); + /** @var RedisFactory $redis */ + $redis = $app->make('redis'); + return new WebhookBatchBuffer( - $app->make('redis')->connection($connectionName) + $redis->connection($connectionName) ); }); From 0af5b7be632d559c166e860c6989dadb2a2bceef Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:38:56 +0000 Subject: [PATCH 02/24] fix: describe Redis multi-exec results precisely Make pipeline and transaction return documentation conditional on whether a callback is supplied. No-callback calls return the native Redis client used to build the batch, while callback calls return the executed result array or false. Add focused coverage for pipeline and transaction callback and no-callback forms, including native false results, without changing runtime dispatch. --- src/redis/src/Traits/MultiExec.php | 8 ++++--- tests/Redis/MultiExecTest.php | 34 ++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 3 deletions(-) diff --git a/src/redis/src/Traits/MultiExec.php b/src/redis/src/Traits/MultiExec.php index 578c98588..088c7ba67 100644 --- a/src/redis/src/Traits/MultiExec.php +++ b/src/redis/src/Traits/MultiExec.php @@ -18,7 +18,7 @@ trait MultiExec /** * Execute commands in a pipeline. * - * @return array|Redis + * @return ($callback is null ? Redis : array|false) */ public function pipeline(?callable $callback = null) { @@ -28,7 +28,7 @@ public function pipeline(?callable $callback = null) /** * Execute commands in a transaction. * - * @return array|Redis|RedisCluster + * @return ($callback is null ? Redis|RedisCluster : array|false) */ public function transaction(?callable $callback = null) { @@ -38,7 +38,9 @@ public function transaction(?callable $callback = null) /** * Execute multi-exec commands with optional callback. * - * @return array|Redis|RedisCluster + * @return ($callback is null + * ? ($command is 'pipeline' ? Redis : Redis|RedisCluster) + * : array|false) */ private function executeMultiExec(string $command, ?callable $callback = null) { diff --git a/tests/Redis/MultiExecTest.php b/tests/Redis/MultiExecTest.php index cb054a281..6bdefd014 100644 --- a/tests/Redis/MultiExecTest.php +++ b/tests/Redis/MultiExecTest.php @@ -67,6 +67,23 @@ public function testPipelineWithCallbackExecutesAndReturnsResults(): void $this->assertSame($execResults, $result); } + public function testPipelineWithCallbackReturnsFalseWhenExecAborts(): void + { + $pipelineInstance = m::mock(PhpRedis::class); + $pipelineInstance->shouldReceive('exec')->once()->andReturnFalse(); + + $phpRedis = m::mock(PhpRedis::class); + $phpRedis->shouldReceive('pipeline')->once()->andReturn($pipelineInstance); + + $connection = $this->createMockConnection($phpRedis); + $connection->shouldReceive('release')->once(); + + $result = $this->createRedis($connection)->pipeline(static function (): void { + }); + + $this->assertFalse($result); + } + public function testTransactionWithoutCallbackReturnsInstanceForChaining(): void { $multiInstance = m::mock(PhpRedis::class); @@ -109,6 +126,23 @@ public function testTransactionWithCallbackExecutesAndReturnsResults(): void $this->assertSame($execResults, $result); } + public function testTransactionWithCallbackReturnsFalseWhenExecAborts(): void + { + $multiInstance = m::mock(PhpRedis::class); + $multiInstance->shouldReceive('exec')->once()->andReturnFalse(); + + $phpRedis = m::mock(PhpRedis::class); + $phpRedis->shouldReceive('multi')->once()->andReturn($multiInstance); + + $connection = $this->createMockConnection($phpRedis); + $connection->shouldReceive('release')->once(); + + $result = $this->createRedis($connection)->transaction(static function (): void { + }); + + $this->assertFalse($result); + } + public function testPipelineWithCallbackDoesNotReleaseExistingContextConnection(): void { $pipelineInstance = m::mock(PhpRedis::class); From 1c645ea710bf95960650a91ac40a0fd55b2fa30f Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:39:02 +0000 Subject: [PATCH 03/24] fix: preserve topology and delays during queue bulk dispatch Use a Redis transaction for Cluster bulk publication and retain the existing pipeline-plus-transaction shape for standalone Redis. This avoids unsupported Cluster pipelines while preserving batched publication semantics. Resolve Delay attributes through the existing cached Queue metadata reader for both Redis and database bulk jobs while retaining public delay properties. Add regressions for both topologies and both delay forms. --- src/queue/src/DatabaseQueue.php | 7 +- src/queue/src/RedisQueue.php | 33 ++++-- tests/Queue/QueueDatabaseQueueUnitTest.php | 30 ++++++ tests/Queue/QueueRedisQueueTest.php | 116 +++++++++++++++++++++ 4 files changed, 175 insertions(+), 11 deletions(-) diff --git a/src/queue/src/DatabaseQueue.php b/src/queue/src/DatabaseQueue.php index bcfe8ce2d..3f56937f7 100644 --- a/src/queue/src/DatabaseQueue.php +++ b/src/queue/src/DatabaseQueue.php @@ -12,6 +12,7 @@ use Hypervel\Database\ConnectionInterface; use Hypervel\Database\ConnectionResolverInterface; use Hypervel\Database\Query\Builder; +use Hypervel\Queue\Attributes\Delay; use Hypervel\Queue\Jobs\DatabaseJob; use Hypervel\Queue\Jobs\DatabaseJobRecord; use Hypervel\Queue\Jobs\InspectedJob; @@ -271,10 +272,14 @@ public function bulk(array $jobs, mixed $data = '', ?string $queue = null): mixe return $this->getDatabase()->table($this->table)->insert(Collection::make((array) $jobs)->map( function ($job) use ($queue, $data, $now) { + $delay = is_object($job) + ? $this->getAttributeValue($job, Delay::class, 'delay') + : null; + return $this->buildDatabaseRecord( $queue, $this->createPayload($job, $this->getQueue($queue), $data), - isset($job->delay) ? $this->availableAt($job->delay) : $now, + $delay !== null ? $this->availableAt($delay) : $now, ); } )->all()); diff --git a/src/queue/src/RedisQueue.php b/src/queue/src/RedisQueue.php index 4a1083545..097952022 100644 --- a/src/queue/src/RedisQueue.php +++ b/src/queue/src/RedisQueue.php @@ -10,6 +10,7 @@ use Hypervel\Contracts\Queue\Job as JobContract; use Hypervel\Contracts\Queue\Queue as QueueContract; use Hypervel\Contracts\Redis\Factory as Redis; +use Hypervel\Queue\Attributes\Delay; use Hypervel\Queue\Jobs\InspectedJob; use Hypervel\Queue\Jobs\RedisJob; use Hypervel\Redis\RedisConnection; @@ -244,17 +245,29 @@ public function creationTimeOfOldestPendingJob(?string $queue = null): ?int */ public function bulk(array $jobs, mixed $data = '', ?string $queue = null): mixed { - $this->getConnection()->pipeline(function () use ($jobs, $data, $queue) { - $this->getConnection()->transaction(function () use ($jobs, $data, $queue) { - foreach ((array) $jobs as $job) { - if (isset($job->delay)) { - $this->later($job->delay, $job, $data, $queue); - } else { - $this->push($job, $data, $queue); - } + $connection = $this->getConnection(); + + $callback = function () use ($jobs, $data, $queue): void { + foreach ($jobs as $job) { + $delay = is_object($job) + ? $this->getAttributeValue($job, Delay::class, 'delay') + : null; + + if ($delay !== null) { + $this->later($delay, $job, $data, $queue); + } else { + $this->push($job, $data, $queue); } - }); - }); + } + }; + + if ($connection->isCluster()) { + $connection->transaction($callback); + } else { + $connection->pipeline( + fn () => $connection->transaction($callback) + ); + } return null; } diff --git a/tests/Queue/QueueDatabaseQueueUnitTest.php b/tests/Queue/QueueDatabaseQueueUnitTest.php index 7d902fa96..8a7168898 100644 --- a/tests/Queue/QueueDatabaseQueueUnitTest.php +++ b/tests/Queue/QueueDatabaseQueueUnitTest.php @@ -11,6 +11,7 @@ use Hypervel\Database\ConnectionInterface; use Hypervel\Database\ConnectionResolverInterface; use Hypervel\Database\Query\Builder; +use Hypervel\Queue\Attributes\Delay; use Hypervel\Queue\DatabaseQueue; use Hypervel\Queue\InvalidPayloadException; use Hypervel\Queue\Jobs\InspectedJob; @@ -226,6 +227,30 @@ public function testBulkBatchPushesOntoDatabase(): void $queue->bulk(['foo', 'bar'], ['data'], 'queue'); } + public function testBulkHonorsTheDelayAttribute(): void + { + CarbonImmutable::setTestNow(CarbonImmutable::createFromTimestamp(1732502704)); + + $queue = new TestDatabaseQueue( + resolver: $resolver = m::mock(ConnectionResolverInterface::class), + connection: null, + table: 'table', + default: 'default', + currentTime: 1732502704, + ); + $resolver->shouldReceive('connection')->andReturn($connection = m::mock(ConnectionInterface::class)); + $connection->shouldReceive('table')->with('table')->andReturn($query = m::mock(Builder::class)); + $query->shouldReceive('insert') + ->once() + ->andReturnUsing(function (array $records): bool { + $this->assertSame(1732502713, $records[0]['available_at']); + + return true; + }); + + $queue->bulk([new DatabaseBulkAttributeDelayJob]); + } + public function testBuildDatabaseRecordWithPayloadAtTheEnd() { $queue = m::mock(DatabaseQueue::class); @@ -458,6 +483,11 @@ class MyBatchableJob use Batchable; } +#[Delay(9)] +class DatabaseBulkAttributeDelayJob +{ +} + class TestDatabaseQueue extends DatabaseQueue { public function __construct( diff --git a/tests/Queue/QueueRedisQueueTest.php b/tests/Queue/QueueRedisQueueTest.php index 9d17fd8b3..19c3427b3 100644 --- a/tests/Queue/QueueRedisQueueTest.php +++ b/tests/Queue/QueueRedisQueueTest.php @@ -4,9 +4,12 @@ namespace Hypervel\Tests\Queue; +use DateInterval; +use DateTimeInterface; use Hypervel\Container\Container; use Hypervel\Contracts\Events\Dispatcher; use Hypervel\Contracts\Redis\Factory as Redis; +use Hypervel\Queue\Attributes\Delay; use Hypervel\Queue\Events\JobQueued; use Hypervel\Queue\Events\JobQueueing; use Hypervel\Queue\Jobs\RedisJob; @@ -18,10 +21,69 @@ use Hypervel\Support\Str; use Hypervel\Tests\TestCase; use Mockery as m; +use Override; use Symfony\Component\Uid\Uuid; class QueueRedisQueueTest extends TestCase { + public function testBulkUsesNestedTransactionOnStandaloneRedisAndHonorsJobDelays(): void + { + $connection = m::mock(RedisProxy::class); + $connection->shouldReceive('isCluster')->once()->andReturnFalse(); + $connection->shouldReceive('pipeline') + ->once() + ->andReturnUsing(static function (callable $callback): array { + $callback(); + + return []; + }); + $connection->shouldReceive('transaction') + ->once() + ->andReturnUsing(static function (callable $callback): array { + $callback(); + + return []; + }); + + $queue = $this->createBulkQueue($connection); + $queue->bulk([ + new RedisBulkPropertyDelayJob, + new RedisBulkAttributeDelayJob, + 'plain', + ], ['data'], 'critical'); + + $this->assertSame([ + [4, RedisBulkPropertyDelayJob::class, ['data'], 'critical'], + [9, RedisBulkAttributeDelayJob::class, ['data'], 'critical'], + ], $queue->delayed); + $this->assertSame([ + ['plain', ['data'], 'critical'], + ], $queue->pushed); + } + + public function testBulkUsesTransactionWithoutPipelineOnRedisCluster(): void + { + $connection = m::mock(RedisProxy::class); + $connection->shouldReceive('isCluster')->once()->andReturnTrue(); + $connection->shouldReceive('transaction') + ->once() + ->andReturnUsing(static function (callable $callback): array { + $callback(); + + return []; + }); + $connection->shouldNotReceive('pipeline'); + + $queue = $this->createBulkQueue($connection); + $queue->bulk(['first', 'second']); + + $this->assertSame([ + ['first', '', null], + ['second', '', null], + ], $queue->pushed); + $this->assertSame([], $queue->delayed); + } + public function testPushProperlyPushesJobOntoRedis(): void { $now = CarbonImmutable::now(); @@ -533,6 +595,14 @@ protected function mockUuid(): Uuid return $uuid; } + + private function createBulkQueue(RedisProxy $connection): BulkTestRedisQueue + { + $redis = m::mock(Redis::class); + $redis->shouldReceive('connection')->once()->with('default')->andReturn($connection); + + return new BulkTestRedisQueue($redis, 'default', 'default'); + } } class TestableRedisQueue extends RedisQueue @@ -547,3 +617,49 @@ public function testIsClusterConnection(): bool return $this->isClusterConnection(); } } + +class BulkTestRedisQueue extends RedisQueue +{ + public array $pushed = []; + + public array $delayed = []; + + #[Override] + public function push(object|string $job, mixed $data = '', ?string $queue = null): mixed + { + $this->pushed[] = [ + is_object($job) ? $job::class : $job, + $data, + $queue, + ]; + + return null; + } + + #[Override] + public function later( + DateInterval|DateTimeInterface|int $delay, + object|string $job, + mixed $data = '', + ?string $queue = null, + ): mixed { + $this->delayed[] = [ + $delay, + is_object($job) ? $job::class : $job, + $data, + $queue, + ]; + + return null; + } +} + +class RedisBulkPropertyDelayJob +{ + public int $delay = 4; +} + +#[Delay(9)] +class RedisBulkAttributeDelayJob +{ +} From 8c4f0ee86f0c5ba5f74bafc49c44c43d7737dac7 Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:39:13 +0000 Subject: [PATCH 04/24] fix: make Horizon repository batches Cluster-aware Route Horizon repository batches through one small topology-aware trait. Standalone connections keep pipelines, while Cluster connections use transactions against the hash-tagged Horizon prefix so every batch remains in one Redis slot. Apply the boundary to command, job, master, process, supervisor, and tag repositories. Add representative topology tests rather than duplicating the same branch assertion at every call site. --- src/horizon/src/RedisHorizonCommandQueue.php | 5 +- .../src/Repositories/RedisJobRepository.php | 20 +++--- .../RedisMasterSupervisorRepository.php | 6 +- .../Repositories/RedisProcessRepository.php | 4 +- .../RedisSupervisorRepository.php | 6 +- .../src/Repositories/RedisTagRepository.php | 8 ++- .../Repositories/UsesClusterAwarePipeline.php | 25 ++++++++ .../Unit/RedisHorizonCommandQueueTest.php | 52 ++++++++++++++++ .../Unit/UsesClusterAwarePipelineTest.php | 61 +++++++++++++++++++ tests/Horizon/UnitTestCase.php | 13 ++++ 10 files changed, 182 insertions(+), 18 deletions(-) create mode 100644 src/horizon/src/Repositories/UsesClusterAwarePipeline.php create mode 100644 tests/Horizon/Unit/RedisHorizonCommandQueueTest.php create mode 100644 tests/Horizon/Unit/UsesClusterAwarePipelineTest.php diff --git a/src/horizon/src/RedisHorizonCommandQueue.php b/src/horizon/src/RedisHorizonCommandQueue.php index 1f9542d74..37b3bab72 100644 --- a/src/horizon/src/RedisHorizonCommandQueue.php +++ b/src/horizon/src/RedisHorizonCommandQueue.php @@ -6,10 +6,13 @@ use Hypervel\Contracts\Redis\Factory as Redis; use Hypervel\Horizon\Contracts\HorizonCommandQueue; +use Hypervel\Horizon\Repositories\UsesClusterAwarePipeline; use Hypervel\Redis\RedisProxy; class RedisHorizonCommandQueue implements HorizonCommandQueue { + use UsesClusterAwarePipeline; + /** * Create a new command queue instance. */ @@ -41,7 +44,7 @@ public function pending(string $name): array return []; } - $results = $this->connection()->pipeline(function ($pipe) use ($name, $length) { + $results = $this->pipeline(function ($pipe) use ($name, $length) { $pipe->lRange('commands:' . $name, 0, $length - 1); $pipe->lTrim('commands:' . $name, $length, -1); diff --git a/src/horizon/src/Repositories/RedisJobRepository.php b/src/horizon/src/Repositories/RedisJobRepository.php index 947c4606a..ca0a71c2e 100644 --- a/src/horizon/src/Repositories/RedisJobRepository.php +++ b/src/horizon/src/Repositories/RedisJobRepository.php @@ -17,6 +17,8 @@ class RedisJobRepository implements JobRepository { + use UsesClusterAwarePipeline; + /** * The keys stored on the job hashes. */ @@ -230,7 +232,7 @@ protected function minutesForType(string $type): int */ public function getJobs(array $ids, mixed $indexFrom = 0): Collection { - $jobs = $this->connection()->pipeline(function ($pipe) use ($ids) { + $jobs = $this->pipeline(function ($pipe) use ($ids) { foreach ($ids as $id) { $pipe->hmget($id, $this->keys); } @@ -264,7 +266,7 @@ protected function indexJobs(Collection $jobs, int $indexFrom): Collection */ public function pushed(string $connection, string $queue, JobPayload $payload): void { - $this->connection()->pipeline(function ($pipe) use ($connection, $queue, $payload) { + $this->pipeline(function ($pipe) use ($connection, $queue, $payload) { $this->storeJobReference($pipe, 'recent_jobs', $payload); $this->storeJobReference($pipe, 'pending_jobs', $payload); @@ -327,7 +329,7 @@ public function released(string $connection, string $queue, JobPayload $payload, */ public function remember(string $connection, string $queue, JobPayload $payload): void { - $this->connection()->pipeline(function ($pipe) use ($connection, $queue, $payload) { + $this->pipeline(function ($pipe) use ($connection, $queue, $payload) { $this->storeJobReference($pipe, 'monitored_jobs', $payload); $pipe->hmset( @@ -355,7 +357,7 @@ public function remember(string $connection, string $queue, JobPayload $payload) */ public function migrated(string $connection, string $queue, Collection $payloads): void { - $this->connection()->pipeline(function ($pipe) use ($payloads) { + $this->pipeline(function ($pipe) use ($payloads) { foreach ($payloads as $payload) { $pipe->hmset( $payload->id(), @@ -379,7 +381,7 @@ public function completed(JobPayload $payload, bool $failed = false, bool $silen $this->updateRetryInformationOnParent($payload, $failed); } - $this->connection()->pipeline(function ($pipe) use ($payload, $silenced) { + $this->pipeline(function ($pipe) use ($payload, $silenced) { $this->storeJobReference($pipe, $silenced ? 'silenced_jobs' : 'completed_jobs', $payload); $this->removeJobReference($pipe, 'pending_jobs', $payload); @@ -432,7 +434,7 @@ protected function updateRetryStatus(JobPayload $payload, array $retries, bool $ */ public function deleteMonitored(array $ids): void { - $this->connection()->pipeline(function ($pipe) use ($ids) { + $this->pipeline(function ($pipe) use ($ids) { foreach ($ids as $id) { $pipe->expireat($id, CarbonImmutable::now()->addDays(7)->getTimestamp()); } @@ -444,7 +446,7 @@ public function deleteMonitored(array $ids): void */ public function trimRecentJobs(): void { - $this->connection()->pipeline(function ($pipe) { + $this->pipeline(function ($pipe) { $pipe->zRemRangeByScore( 'recent_jobs', (string) (CarbonImmutable::now()->subMinutes($this->recentJobExpires)->getTimestamp() * -1), @@ -527,7 +529,7 @@ public function findFailed(string $id): ?stdClass */ public function failed(Throwable $exception, string $connection, string $queue, JobPayload $payload): void { - $this->connection()->pipeline(function ($pipe) use ($exception, $connection, $queue, $payload) { + $this->pipeline(function ($pipe) use ($exception, $connection, $queue, $payload) { $this->storeJobReference($pipe, 'failed_jobs', $payload); $this->storeJobReference($pipe, 'recent_failed_jobs', $payload); $this->removeJobReference($pipe, 'pending_jobs', $payload); @@ -615,7 +617,7 @@ public function purge(string $queue): int 2, 'recent_jobs', 'pending_jobs', - config('horizon.prefix'), + config()->string('horizon.prefix'), $queue, $cursor, ); diff --git a/src/horizon/src/Repositories/RedisMasterSupervisorRepository.php b/src/horizon/src/Repositories/RedisMasterSupervisorRepository.php index 122adde9a..23afacdbc 100644 --- a/src/horizon/src/Repositories/RedisMasterSupervisorRepository.php +++ b/src/horizon/src/Repositories/RedisMasterSupervisorRepository.php @@ -15,6 +15,8 @@ class RedisMasterSupervisorRepository implements MasterSupervisorRepository { + use UsesClusterAwarePipeline; + /** * Create a new repository instance. */ @@ -56,7 +58,7 @@ public function find(string $name): ?stdClass */ public function get(array $names): array { - $records = $this->connection()->pipeline(function ($pipe) use ($names) { + $records = $this->pipeline(function ($pipe) use ($names) { foreach ($names as $name) { $pipe->hmget('master:' . $name, ['name', 'pid', 'status', 'supervisors', 'environment']); } @@ -77,7 +79,7 @@ public function update(MasterSupervisor $master): void /** @phpstan-ignore-next-line */ $supervisors = $master->supervisors->map->name->all(); - $this->connection()->pipeline(function ($pipe) use ($master, $supervisors) { + $this->pipeline(function ($pipe) use ($master, $supervisors) { $pipe->hmset( 'master:' . $master->name, [ diff --git a/src/horizon/src/Repositories/RedisProcessRepository.php b/src/horizon/src/Repositories/RedisProcessRepository.php index eb610c8a8..3fbd8f414 100644 --- a/src/horizon/src/Repositories/RedisProcessRepository.php +++ b/src/horizon/src/Repositories/RedisProcessRepository.php @@ -11,6 +11,8 @@ class RedisProcessRepository implements ProcessRepository { + use UsesClusterAwarePipeline; + /** * Create a new repository instance. * @@ -46,7 +48,7 @@ public function orphaned(string $master, array $processIds): void $this->connection()->hDel($key, ...$shouldRemove); } - $this->connection()->pipeline(function ($pipe) use ($key, $time, $processIds) { + $this->pipeline(function ($pipe) use ($key, $time, $processIds) { foreach ($processIds as $processId) { $pipe->hSetNx($key, $processId, $time); } diff --git a/src/horizon/src/Repositories/RedisSupervisorRepository.php b/src/horizon/src/Repositories/RedisSupervisorRepository.php index 41421f3f9..5582ef3ce 100644 --- a/src/horizon/src/Repositories/RedisSupervisorRepository.php +++ b/src/horizon/src/Repositories/RedisSupervisorRepository.php @@ -14,6 +14,8 @@ class RedisSupervisorRepository implements SupervisorRepository { + use UsesClusterAwarePipeline; + /** * Create a new repository instance. */ @@ -56,7 +58,7 @@ public function find(string $name): ?stdClass public function get(array $names): array { /** @var array $records */ - $records = $this->connection()->pipeline(function ($pipe) use ($names) { + $records = $this->pipeline(function ($pipe) use ($names) { foreach ($names as $name) { $pipe->hmget('supervisor:' . $name, ['name', 'master', 'pid', 'status', 'processes', 'options']); } @@ -91,7 +93,7 @@ public function update(Supervisor $supervisor): void return [$supervisor->options->connection . ':' . $pool->queue() => count($pool->processes())]; })->toJson(); - $this->connection()->pipeline(function ($pipe) use ($supervisor, $processes) { + $this->pipeline(function ($pipe) use ($supervisor, $processes) { $pipe->hmset( 'supervisor:' . $supervisor->name, [ diff --git a/src/horizon/src/Repositories/RedisTagRepository.php b/src/horizon/src/Repositories/RedisTagRepository.php index c9b293673..4c47e802c 100644 --- a/src/horizon/src/Repositories/RedisTagRepository.php +++ b/src/horizon/src/Repositories/RedisTagRepository.php @@ -10,6 +10,8 @@ class RedisTagRepository implements TagRepository { + use UsesClusterAwarePipeline; + /** * Create a new repository instance. * @@ -57,7 +59,7 @@ public function stopMonitoring(string $tag): void */ public function add(string $id, array $tags): void { - $this->connection()->pipeline(function ($pipe) use ($id, $tags) { + $this->pipeline(function ($pipe) use ($id, $tags) { foreach ($tags as $tag) { $pipe->zAdd($tag, str_replace(',', '.', (string) microtime(true)), $id); } @@ -69,7 +71,7 @@ public function add(string $id, array $tags): void */ public function addTemporary(int $minutes, string $id, array $tags): void { - $this->connection()->pipeline(function ($pipe) use ($minutes, $id, $tags) { + $this->pipeline(function ($pipe) use ($minutes, $id, $tags) { foreach ($tags as $tag) { $pipe->zAdd($tag, str_replace(',', '.', (string) microtime(true)), $id); @@ -115,7 +117,7 @@ public function paginate(string $tag, int $startingAt = 0, int $limit = 25): arr */ public function forgetJobs(array|string $tags, array|string $ids): void { - $this->connection()->pipeline(function ($pipe) use ($tags, $ids) { + $this->pipeline(function ($pipe) use ($tags, $ids) { foreach ((array) $tags as $tag) { foreach ((array) $ids as $id) { $pipe->zrem($tag, $id); diff --git a/src/horizon/src/Repositories/UsesClusterAwarePipeline.php b/src/horizon/src/Repositories/UsesClusterAwarePipeline.php new file mode 100644 index 000000000..f02d3f238 --- /dev/null +++ b/src/horizon/src/Repositories/UsesClusterAwarePipeline.php @@ -0,0 +1,25 @@ +connection(); + + // Horizon hash-tags its Cluster prefix, so this remains one atomic + // single-slot batch. Horizon never uses WATCH, so EXEC cannot abort. + /** @var array $result */ + $result = $connection->isCluster() + ? $connection->transaction($callback) + : $connection->pipeline($callback); + + return $result; + } +} diff --git a/tests/Horizon/Unit/RedisHorizonCommandQueueTest.php b/tests/Horizon/Unit/RedisHorizonCommandQueueTest.php new file mode 100644 index 000000000..4724053cd --- /dev/null +++ b/tests/Horizon/Unit/RedisHorizonCommandQueueTest.php @@ -0,0 +1,52 @@ +createQueue(cluster: false, method: 'pipeline'); + + $this->assertSame('pause', $queue->pending('master')[0]->command); + } + + public function testPendingUsesTransactionOnRedisCluster(): void + { + $queue = $this->createQueue(cluster: true, method: 'transaction'); + + $this->assertSame('pause', $queue->pending('master')[0]->command); + } + + private function createQueue(bool $cluster, string $method): RedisHorizonCommandQueue + { + $pipeline = m::mock(); + $pipeline->shouldReceive('lRange')->once()->with('commands:master', 0, 0); + $pipeline->shouldReceive('lTrim')->once()->with('commands:master', 1, -1); + + $connection = m::mock(RedisProxy::class); + $connection->shouldReceive('lLen')->once()->with('commands:master')->andReturn(1); + $connection->shouldReceive('isCluster')->once()->andReturn($cluster); + $connection->shouldReceive($method) + ->once() + ->andReturnUsing(function (callable $callback) use ($pipeline): array { + $callback($pipeline); + + return [[json_encode(['command' => 'pause', 'options' => []])], true]; + }); + $connection->shouldNotReceive($method === 'pipeline' ? 'transaction' : 'pipeline'); + + $redis = m::mock(Factory::class); + $redis->shouldReceive('connection')->twice()->with('horizon')->andReturn($connection); + + return new RedisHorizonCommandQueue($redis); + } +} diff --git a/tests/Horizon/Unit/UsesClusterAwarePipelineTest.php b/tests/Horizon/Unit/UsesClusterAwarePipelineTest.php new file mode 100644 index 000000000..29048fcd0 --- /dev/null +++ b/tests/Horizon/Unit/UsesClusterAwarePipelineTest.php @@ -0,0 +1,61 @@ +shouldReceive('isCluster')->once()->andReturnFalse(); + $connection->shouldReceive('pipeline')->once()->andReturn(['pipeline']); + $connection->shouldNotReceive('transaction'); + + $this->assertSame( + ['pipeline'], + (new ClusterAwarePipelineHarness($connection))->execute(static function (): void { + }), + ); + } + + public function testClusterConnectionUsesTransaction(): void + { + $connection = m::mock(RedisProxy::class); + $connection->shouldReceive('isCluster')->once()->andReturnTrue(); + $connection->shouldReceive('transaction')->once()->andReturn(['transaction']); + $connection->shouldNotReceive('pipeline'); + + $this->assertSame( + ['transaction'], + (new ClusterAwarePipelineHarness($connection))->execute(static function (): void { + }), + ); + } +} + +class ClusterAwarePipelineHarness +{ + use UsesClusterAwarePipeline; + + public function __construct( + private RedisProxy $redis + ) { + } + + public function execute(callable $callback): array + { + return $this->pipeline($callback); + } + + protected function connection(): RedisProxy + { + return $this->redis; + } +} diff --git a/tests/Horizon/UnitTestCase.php b/tests/Horizon/UnitTestCase.php index d30d832a2..4b8d375fd 100644 --- a/tests/Horizon/UnitTestCase.php +++ b/tests/Horizon/UnitTestCase.php @@ -4,8 +4,21 @@ namespace Hypervel\Tests\Horizon; +use Hypervel\Contracts\Redis\Factory; +use Hypervel\Redis\RedisProxy; use Hypervel\Tests\TestCase; +use Mockery as m; abstract class UnitTestCase extends TestCase { + /** + * Create a Redis factory for the given Horizon connection. + */ + protected function redisFactory(RedisProxy $connection): Factory + { + $redis = m::mock(Factory::class); + $redis->shouldReceive('connection')->once()->with('horizon')->andReturn($connection); + + return $redis; + } } From a49a54b2c1f65e78e89d614dc8a0a5bbb5f52ef6 Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:39:21 +0000 Subject: [PATCH 05/24] fix: consume Horizon queue telemetry context exactly once Consume and forget the last-pushed job before preparing raw payloads, and clear listener-event context in a finally block. Later work in the same coroutine can no longer inherit stale telemetry after success or failure. Use the physical hash-tag-aware queue key for ready counts and include the resolved queue and delay in delayed payload creation. Add coverage for direct raw pushes, preparation failures, delayed metadata, hook queue names, and listener tag failures. --- src/horizon/src/RedisQueue.php | 19 +++-- src/horizon/src/Tags.php | 19 +++-- tests/Horizon/Unit/RedisQueueTest.php | 28 ++++++++ .../Listeners/StoreTagsForFailedTest.php | 19 +++++ .../Horizon/Feature/QueueProcessingTest.php | 71 ++++++++++++++++++- .../Horizon/Feature/RedisPayloadTest.php | 34 +++++++++ 6 files changed, 171 insertions(+), 19 deletions(-) create mode 100644 tests/Horizon/Unit/RedisQueueTest.php diff --git a/src/horizon/src/RedisQueue.php b/src/horizon/src/RedisQueue.php index a6be881d6..0a7509738 100644 --- a/src/horizon/src/RedisQueue.php +++ b/src/horizon/src/RedisQueue.php @@ -30,7 +30,7 @@ class RedisQueue extends BaseQueue */ public function readyNow(?string $queue = null): int { - return $this->getConnection()->lLen($this->getQueue($queue)); + return $this->getConnection()->lLen($this->getQueueRedisKey($queue)); } /** @@ -59,7 +59,10 @@ static function (BaseQueue $owner, $payload, $queue) use ($job) { #[Override] public function pushRaw(string $payload, ?string $queue = null, array $options = []): mixed { - $payload = (new JobPayload($payload))->prepare($this->getLastPushed()); + $job = CoroutineContext::get(static::LAST_PUSHED_CONTEXT_KEY); + CoroutineContext::forget(static::LAST_PUSHED_CONTEXT_KEY); + + $payload = (new JobPayload($payload))->prepare($job); $this->event($this->getQueue($queue), new JobPending($payload->value)); @@ -89,7 +92,9 @@ protected function createPayloadArray(array|object|string $job, ?string $queue, #[Override] public function later(DateInterval|DateTimeInterface|int $delay, object|string $job, mixed $data = '', ?string $queue = null): mixed { - $payload = (new JobPayload($this->createPayload($job, $queue, $data)))->prepare($job)->value; + $payload = (new JobPayload( + $this->createPayload($job, $this->getQueue($queue), $data, $delay) + ))->prepare($job)->value; return $this->enqueueUsing( $job, @@ -194,12 +199,4 @@ protected function setLastPushed(object|string $job): void { CoroutineContext::set(static::LAST_PUSHED_CONTEXT_KEY, $job); } - - /** - * Get the job that last pushed to queue via the "push" method. - */ - protected function getLastPushed(): object|string|null - { - return CoroutineContext::get(static::LAST_PUSHED_CONTEXT_KEY); - } } diff --git a/src/horizon/src/Tags.php b/src/horizon/src/Tags.php index 8ef34e9be..c1b876d5b 100644 --- a/src/horizon/src/Tags.php +++ b/src/horizon/src/Tags.php @@ -53,13 +53,15 @@ protected static function tagsForListener(CallQueuedListener $job): array static::setEvent($event); - return collect( - [static::extractListener($job), $event] - )->map(function ($job) { - return static::for($job); - })->collapse()->unique()->tap(function () { + try { + return collect([static::extractListener($job), $event]) + ->map(fn ($job) => static::for($job)) + ->collapse() + ->unique() + ->toArray(); + } finally { static::flushEventState(); - })->toArray(); + } } /** @@ -151,6 +153,9 @@ protected static function setEvent(object $event): void CoroutineContext::set(static::CONTEXT_KEY, $event); } + /** + * Get the event currently being handled. + */ protected static function getEvent(): ?object { return CoroutineContext::get(static::CONTEXT_KEY); @@ -161,6 +166,6 @@ protected static function getEvent(): ?object */ protected static function flushEventState(): void { - CoroutineContext::set(static::CONTEXT_KEY, null); + CoroutineContext::forget(static::CONTEXT_KEY); } } diff --git a/tests/Horizon/Unit/RedisQueueTest.php b/tests/Horizon/Unit/RedisQueueTest.php new file mode 100644 index 000000000..f2368ebbb --- /dev/null +++ b/tests/Horizon/Unit/RedisQueueTest.php @@ -0,0 +1,28 @@ +shouldReceive('isCluster')->once()->andReturnTrue(); + $connection->shouldReceive('lLen')->once()->with('queues:{critical}')->andReturn(4); + + $redis = m::mock(Factory::class); + $redis->shouldReceive('connection')->twice()->with('default')->andReturn($connection); + + $queue = new RedisQueue($redis, 'default', 'default'); + + $this->assertSame(4, $queue->readyNow('critical')); + } +} diff --git a/tests/Integration/Horizon/Feature/Listeners/StoreTagsForFailedTest.php b/tests/Integration/Horizon/Feature/Listeners/StoreTagsForFailedTest.php index db21e93fa..1d7dc81ec 100644 --- a/tests/Integration/Horizon/Feature/Listeners/StoreTagsForFailedTest.php +++ b/tests/Integration/Horizon/Feature/Listeners/StoreTagsForFailedTest.php @@ -33,6 +33,25 @@ public function testTemporaryFailedJobShouldBeDeletedWhenTheMainJobIsDeleted(): $this->app->make(Dispatcher::class)->dispatch($event); } + + public function testFailedJobTrimDefaultSurvivesReplaceWholeConfiguration(): void + { + config()->set('horizon.trim', ['recent' => 60]); + + $tagRepository = m::mock(TagRepository::class); + $tagRepository->shouldReceive('addTemporary')->once()->with(10080, '1', ['failed:foobar'])->andReturn([]); + + $this->instance(TagRepository::class, $tagRepository); + + $event = new JobFailed( + new Exception('job failed'), + new FailedJob, + '{"id":"1","displayName":"displayName","tags":["foobar"]}' + ); + $event->connection('redis')->queue('default'); + + $this->app->make(Dispatcher::class)->dispatch($event); + } } class FailedJob extends Job diff --git a/tests/Integration/Horizon/Feature/QueueProcessingTest.php b/tests/Integration/Horizon/Feature/QueueProcessingTest.php index f32b99fc6..fcdd300ca 100644 --- a/tests/Integration/Horizon/Feature/QueueProcessingTest.php +++ b/tests/Integration/Horizon/Feature/QueueProcessingTest.php @@ -7,6 +7,9 @@ use Hypervel\Horizon\Contracts\JobRepository; use Hypervel\Horizon\Events\JobReserved; use Hypervel\Horizon\Events\JobsMigrated; +use Hypervel\Horizon\RedisQueue; +use Hypervel\Queue\InvalidPayloadException; +use Hypervel\Queue\Queue as BaseQueue; use Hypervel\Support\CarbonImmutable; use Hypervel\Support\Facades\Event; use Hypervel\Support\Facades\Queue; @@ -36,11 +39,69 @@ public function testPendingJobsAreStoredInPendingJobDatabase() $this->assertSame('pending', Redis::connection('horizon')->hget($id, 'status')); } - public function testPendingDelayedJobsAreStoredInPendingJobDatabase() + public function testPendingDelayedJobsAreStoredInPendingJobDatabase(): void { $id = Queue::later(1, new Jobs\BasicJob); $this->assertSame(1, $this->recentJobs()); $this->assertSame('pending', Redis::connection('horizon')->hget($id, 'status')); + + $payload = json_decode(Redis::connection('horizon')->hget($id, 'payload'), true); + $this->assertSame(1, $payload['delay']); + } + + public function testImmediateAndDelayedPayloadHooksReceiveTheResolvedQueue(): void + { + $queues = []; + BaseQueue::createPayloadUsing(function (string $connection, string $queue) use (&$queues): array { + $queues[] = $queue; + + return []; + }); + + try { + /** @var RedisQueue $queue */ + $queue = Queue::connection('redis'); + $queue->push(new Jobs\BasicJob, queue: 'critical'); + $queue->later(1, new Jobs\BasicJob, queue: 'critical'); + } finally { + BaseQueue::createPayloadUsing(null); + } + + $this->assertSame(['queues:critical', 'queues:critical'], $queues); + } + + public function testDirectRawPushDoesNotInheritThePreviousJob(): void + { + Queue::push(new Jobs\BasicJob); + + /** @var RedisQueue $queue */ + $queue = Queue::connection('redis'); + $queue->pushRaw('{"id":"raw-id","displayName":"Raw Job"}'); + + $payload = json_decode(Redis::connection('horizon')->hget('raw-id', 'payload'), true); + $this->assertSame([], $payload['tags']); + } + + public function testPayloadPreparationFailureStillConsumesThePreviousJob(): void + { + $queue = new RedisQueueWithExposedLastPushed( + app('redis'), + 'default', + 'default', + ); + $queue->setContainer($this->app)->setConnectionName('redis'); + $queue->rememberLastPushed(new Jobs\BasicJob); + + try { + $queue->pushRaw('{invalid'); + $this->fail('Expected the invalid payload to be rejected.'); + } catch (InvalidPayloadException) { + } + + $queue->pushRaw('{"id":"raw-after-failure","displayName":"Raw Job"}'); + + $payload = json_decode(Redis::connection('horizon')->hget('raw-after-failure', 'payload'), true); + $this->assertSame([], $payload['tags']); } public function testPendingJobsAreStoredWithTheirTags() @@ -120,3 +181,11 @@ public function testMigratedTelemetryRetainsValidPayloadsFromMixedInput(): void $this->assertSame('valid', $event->payloads->first()->id()); } } + +class RedisQueueWithExposedLastPushed extends RedisQueue +{ + public function rememberLastPushed(object|string $job): void + { + $this->setLastPushed($job); + } +} diff --git a/tests/Integration/Horizon/Feature/RedisPayloadTest.php b/tests/Integration/Horizon/Feature/RedisPayloadTest.php index f4ba44903..3b860ee4d 100644 --- a/tests/Integration/Horizon/Feature/RedisPayloadTest.php +++ b/tests/Integration/Horizon/Feature/RedisPayloadTest.php @@ -10,6 +10,7 @@ use Hypervel\Events\CallQueuedListener; use Hypervel\Horizon\Contracts\Silenced; use Hypervel\Horizon\JobPayload; +use Hypervel\Horizon\Tags; use Hypervel\Mail\SendQueuedMailable; use Hypervel\Notifications\SendQueuedNotifications; use Hypervel\Queue\InvalidPayloadException; @@ -28,6 +29,7 @@ use Hypervel\Tests\Integration\Horizon\Feature\Fixtures\SilencedMailable; use Hypervel\Tests\Integration\Horizon\IntegrationTestCase; use Mockery as m; +use RuntimeException; use StdClass; class RedisPayloadTest extends IntegrationTestCase @@ -219,4 +221,36 @@ public function testPayloadRequiresAnArrayWithAStringIdentifier(): void } } } + + public function testListenerTagFailureDoesNotLeakItsEventIntoLaterTagExtraction(): void + { + try { + Tags::for(new CallQueuedListener( + FailingListenerTags::class, + 'handle', + [new FakeEvent], + )); + $this->fail('Expected listener tag extraction to fail.'); + } catch (RuntimeException $exception) { + $this->assertSame('Unable to extract listener tags.', $exception->getMessage()); + } + + $this->assertSame(['no-event'], Tags::for(new ListenerEventStateInspectingJob)); + } +} + +class FailingListenerTags +{ + public function tags(FakeEvent $event): array + { + throw new RuntimeException('Unable to extract listener tags.'); + } +} + +class ListenerEventStateInspectingJob +{ + public function tags(?object $event): array + { + return [$event === null ? 'no-event' : 'stale-event']; + } } From 6fb1de00a2765b960d7603c78d446a328fc8029e Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:39:30 +0000 Subject: [PATCH 06/24] fix: correct Horizon metrics ownership and clearing Read the actual maximum ranking entry, keep native snapshot field types truthful, and release stopwatch entries from the listener that consumes them even when either metric write fails. Clear all metrics through one raw Redis lease and the existing prefix-aware scan primitive. Make snapshot lock duration configuration explicit while retaining the nested-array fallback required by shallow configuration replacement. --- src/horizon/src/Console/SnapshotCommand.php | 4 +- .../src/Listeners/UpdateJobMetrics.php | 24 ++++--- .../Repositories/RedisMetricsRepository.php | 40 +++++------ tests/Horizon/Console/SnapshotCommandTest.php | 49 +++++++++++++ .../Unit/RedisMetricsRepositoryTest.php | 63 +++++++++++++++++ tests/Horizon/Unit/UpdateJobMetricsTest.php | 70 +++++++++++++++++++ .../Horizon/Feature/MetricsTest.php | 34 +++++---- 7 files changed, 236 insertions(+), 48 deletions(-) create mode 100644 tests/Horizon/Console/SnapshotCommandTest.php create mode 100644 tests/Horizon/Unit/RedisMetricsRepositoryTest.php create mode 100644 tests/Horizon/Unit/UpdateJobMetricsTest.php diff --git a/src/horizon/src/Console/SnapshotCommand.php b/src/horizon/src/Console/SnapshotCommand.php index b17b1e376..76f97050e 100644 --- a/src/horizon/src/Console/SnapshotCommand.php +++ b/src/horizon/src/Console/SnapshotCommand.php @@ -27,7 +27,9 @@ class SnapshotCommand extends Command */ public function handle(Lock $lock, MetricsRepository $metrics): void { - if ($lock->get('metrics:snapshot', config('horizon.metrics.snapshot_lock', 300))) { + $seconds = config()->integer('horizon.metrics.snapshot_lock', 300) - 30; + + if ($lock->get('metrics:snapshot', $seconds)) { $metrics->snapshot(); $this->components->info('Metrics snapshot stored successfully.'); diff --git a/src/horizon/src/Listeners/UpdateJobMetrics.php b/src/horizon/src/Listeners/UpdateJobMetrics.php index b378719eb..33994cb14 100644 --- a/src/horizon/src/Listeners/UpdateJobMetrics.php +++ b/src/horizon/src/Listeners/UpdateJobMetrics.php @@ -33,16 +33,18 @@ public function handle(JobDeleted $event): void $time = $this->watch->check($id = $event->payload->id()) ?: 0; - $this->metrics->incrementQueue( - $event->job->getQueue(), - $time - ); - - $this->metrics->incrementJob( - $event->payload->displayName(), - $time - ); - - $this->watch->forget($id); + try { + $this->metrics->incrementQueue( + $event->job->getQueue(), + $time + ); + + $this->metrics->incrementJob( + $event->payload->displayName(), + $time + ); + } finally { + $this->watch->forget($id); + } } } diff --git a/src/horizon/src/Repositories/RedisMetricsRepository.php b/src/horizon/src/Repositories/RedisMetricsRepository.php index 36c8b02a6..f0c9cfc83 100644 --- a/src/horizon/src/Repositories/RedisMetricsRepository.php +++ b/src/horizon/src/Repositories/RedisMetricsRepository.php @@ -9,10 +9,9 @@ use Hypervel\Horizon\Lock; use Hypervel\Horizon\LuaScripts; use Hypervel\Horizon\WaitTimeCalculator; -use Hypervel\Redis\PhpRedis; +use Hypervel\Redis\RedisConnection; use Hypervel\Redis\RedisProxy; use Hypervel\Support\CarbonImmutable; -use Hypervel\Support\Str; class RedisMetricsRepository implements MetricsRepository { @@ -120,7 +119,7 @@ protected function runtimeFor(string $key): float public function queueWithMaximumRuntime(): ?string { return collect($this->measuredQueues())->sortBy(function ($queue) { - if ($snapshots = $this->connection()->zRange('snapshot:queue:' . $queue, -1, 1)) { + if ($snapshots = $this->connection()->zRange('snapshot:queue:' . $queue, -1, -1)) { return json_decode($snapshots[0])->runtime; } })->last(); @@ -132,7 +131,7 @@ public function queueWithMaximumRuntime(): ?string public function queueWithMaximumThroughput(): ?string { return collect($this->measuredQueues())->sortBy(function ($queue) { - if ($snapshots = $this->connection()->zRange('snapshot:queue:' . $queue, -1, 1)) { + if ($snapshots = $this->connection()->zRange('snapshot:queue:' . $queue, -1, -1)) { return json_decode($snapshots[0])->throughput; } })->last(); @@ -261,11 +260,12 @@ protected function storeSnapshotForQueue(string $queue): void /** * Get the base snapshot data for a given key. * - * @return array{throughput: string, runtime: string} + * @return array{throughput: false|string, runtime: false|string} */ protected function baseSnapshotData(string $key): array { - /** @var array{0: array{throughput: string, runtime: string}} $responses */ + // Horizon never issues WATCH, so EXEC cannot abort this transaction. + /** @var array{0: array{throughput: false|string, runtime: false|string}} $responses */ $responses = $this->connection()->transaction(function ($trans) use ($key) { $trans->hmget($key, ['throughput', 'runtime']); @@ -320,25 +320,21 @@ public function forget(string $key): void */ public function clear(): void { - $this->forget('last_snapshot_at'); - $this->forget('measured_jobs'); - $this->forget('measured_queues'); - $this->forget('metrics:snapshot'); - - foreach (['queue:*', 'job:*', 'snapshot:*'] as $pattern) { - $cursor = PhpRedis::initialScanCursor(); - - do { - [$cursor, $keys] = $this->connection()->scan( - $cursor, - ['match' => config('horizon.prefix') . $pattern] + $this->connection()->withConnection( + function (RedisConnection $connection): void { + $connection->del( + 'last_snapshot_at', + 'measured_jobs', + 'measured_queues', + 'metrics:snapshot', ); - foreach ($keys ?? [] as $key) { - $this->forget(Str::after($key, config('horizon.prefix'))); + foreach (['queue:*', 'job:*', 'snapshot:*'] as $pattern) { + $connection->flushByPattern($pattern); } - } while ($cursor > 0); - } + }, + transform: false, + ); } /** diff --git a/tests/Horizon/Console/SnapshotCommandTest.php b/tests/Horizon/Console/SnapshotCommandTest.php new file mode 100644 index 000000000..254038b04 --- /dev/null +++ b/tests/Horizon/Console/SnapshotCommandTest.php @@ -0,0 +1,49 @@ +shouldReceive('get')->once()->with('metrics:snapshot', 270)->andReturnFalse(); + + $this->app->make(SnapshotCommand::class)->handle( + $lock, + m::mock(MetricsRepository::class), + ); + } + + public function testSnapshotLockDefaultSurvivesReplaceWholeMetricsConfiguration(): void + { + config(['horizon.metrics' => [ + 'trim_snapshots' => ['job' => 12, 'queue' => 12], + ]]); + + $lock = m::mock(Lock::class); + $lock->shouldReceive('get')->once()->with('metrics:snapshot', 270)->andReturnFalse(); + + $this->app->make(SnapshotCommand::class)->handle( + $lock, + m::mock(MetricsRepository::class), + ); + } +} diff --git a/tests/Horizon/Unit/RedisMetricsRepositoryTest.php b/tests/Horizon/Unit/RedisMetricsRepositoryTest.php new file mode 100644 index 000000000..4c3c957ec --- /dev/null +++ b/tests/Horizon/Unit/RedisMetricsRepositoryTest.php @@ -0,0 +1,63 @@ +shouldReceive('del') + ->once() + ->with('last_snapshot_at', 'measured_jobs', 'measured_queues', 'metrics:snapshot'); + $rawConnection->shouldReceive('flushByPattern')->once()->with('queue:*'); + $rawConnection->shouldReceive('flushByPattern')->once()->with('job:*'); + $rawConnection->shouldReceive('flushByPattern')->once()->with('snapshot:*'); + + $connection = m::mock(RedisProxy::class); + $connection->shouldReceive('withConnection') + ->once() + ->withArgs(function (callable $callback, bool $transform) use ($rawConnection): bool { + $this->assertFalse($transform); + $callback($rawConnection); + + return true; + }); + + (new RedisMetricsRepository($this->redisFactory($connection)))->clear(); + } + + public function testMissingSnapshotHashFieldsRemainFalse(): void + { + $connection = m::mock(RedisProxy::class); + $connection->shouldReceive('transaction') + ->once() + ->andReturn([[ + 'throughput' => false, + 'runtime' => false, + ], 0]); + + $repository = new TestRedisMetricsRepository($this->redisFactory($connection)); + + $this->assertSame([ + 'throughput' => false, + 'runtime' => false, + ], $repository->baseSnapshotDataForTest('queue:default')); + } +} + +class TestRedisMetricsRepository extends RedisMetricsRepository +{ + public function baseSnapshotDataForTest(string $key): array + { + return $this->baseSnapshotData($key); + } +} diff --git a/tests/Horizon/Unit/UpdateJobMetricsTest.php b/tests/Horizon/Unit/UpdateJobMetricsTest.php new file mode 100644 index 000000000..36b8eb096 --- /dev/null +++ b/tests/Horizon/Unit/UpdateJobMetricsTest.php @@ -0,0 +1,70 @@ +shouldReceive('incrementQueue')->once()->with('critical', 1.5)->andThrow($exception); + $metrics->shouldNotReceive('incrementJob'); + + $watch = m::mock(Stopwatch::class); + $watch->shouldReceive('check')->once()->with('job-id')->andReturn(1.5); + $watch->shouldReceive('forget')->once()->with('job-id'); + + try { + (new UpdateJobMetrics($metrics, $watch))->handle($this->event()); + $this->fail('Expected the metrics failure to propagate.'); + } catch (RuntimeException $thrown) { + $this->assertSame($exception, $thrown); + } + } + + public function testJobMetricFailureStillReleasesTheStopwatchEntry(): void + { + $exception = new RuntimeException('job metrics failed'); + $metrics = m::mock(MetricsRepository::class); + $metrics->shouldReceive('incrementQueue')->once()->with('critical', 1.5); + $metrics->shouldReceive('incrementJob')->once()->with('ExampleJob', 1.5)->andThrow($exception); + + $watch = m::mock(Stopwatch::class); + $watch->shouldReceive('check')->once()->with('job-id')->andReturn(1.5); + $watch->shouldReceive('forget')->once()->with('job-id'); + + try { + (new UpdateJobMetrics($metrics, $watch))->handle($this->event()); + $this->fail('Expected the metrics failure to propagate.'); + } catch (RuntimeException $thrown) { + $this->assertSame($exception, $thrown); + } + } + + private function event(): JobDeleted + { + $job = m::mock(RedisJob::class); + $job->shouldReceive('hasFailed')->once()->andReturnFalse(); + $job->shouldReceive('getQueue')->once()->andReturn('critical'); + + return new JobDeleted( + $job, + json_encode([ + 'id' => 'job-id', + 'displayName' => 'ExampleJob', + ]), + ); + } +} diff --git a/tests/Integration/Horizon/Feature/MetricsTest.php b/tests/Integration/Horizon/Feature/MetricsTest.php index af5667c22..c53ff0d1c 100644 --- a/tests/Integration/Horizon/Feature/MetricsTest.php +++ b/tests/Integration/Horizon/Feature/MetricsTest.php @@ -295,34 +295,40 @@ public function testClearIsIdempotent() $this->assertSame(0, $metrics->throughput()); } - public function testQueueWithMaximumRuntime() + public function testQueueWithMaximumRuntime(): void { $metrics = resolve(MetricsRepository::class); - // Set up metrics for two queues with different runtimes - $metrics->incrementQueue('fast-queue', 100.0); - $metrics->incrementQueue('slow-queue', 500.0); - CarbonImmutable::setTestNow(CarbonImmutable::now()); - $metrics->snapshot(); + + // Multiple snapshots expose the old range; the z-prefix makes its fallback choose incorrectly. + for ($snapshot = 0; $snapshot < 3; ++$snapshot) { + $metrics->incrementQueue('z-fast-queue', 100.0); + $metrics->incrementQueue('slow-queue', 500.0); + $metrics->snapshot(); + CarbonImmutable::setTestNow(CarbonImmutable::now()->addSecond()); + } $this->assertSame('slow-queue', $metrics->queueWithMaximumRuntime()); CarbonImmutable::setTestNow(); } - public function testQueueWithMaximumThroughput() + public function testQueueWithMaximumThroughput(): void { $metrics = resolve(MetricsRepository::class); - // Set up metrics — busy queue gets 3 jobs, quiet queue gets 1 - $metrics->incrementQueue('busy-queue', 100.0); - $metrics->incrementQueue('busy-queue', 100.0); - $metrics->incrementQueue('busy-queue', 100.0); - $metrics->incrementQueue('quiet-queue', 100.0); - CarbonImmutable::setTestNow(CarbonImmutable::now()); - $metrics->snapshot(); + + // Multiple snapshots expose the old range; the z-prefix makes its fallback choose incorrectly. + for ($snapshot = 0; $snapshot < 3; ++$snapshot) { + $metrics->incrementQueue('busy-queue', 100.0); + $metrics->incrementQueue('busy-queue', 100.0); + $metrics->incrementQueue('busy-queue', 100.0); + $metrics->incrementQueue('z-quiet-queue', 100.0); + $metrics->snapshot(); + CarbonImmutable::setTestNow(CarbonImmutable::now()->addSecond()); + } $this->assertSame('busy-queue', $metrics->queueWithMaximumThroughput()); From cb3aaf59709cc3c4f763c805ffa93ed5fb33a465 Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:39:39 +0000 Subject: [PATCH 07/24] fix: make Horizon locks atomic and owner-safe Acquire ordinary Horizon locks with one atomic SET EX NX operation and reject non-positive lifetimes before connecting. Callback locks delegate to the existing token-owned RedisLock implementation so an expired owner cannot delete its successor. Preserve the public force-release behavior and primary callback exceptions. Add focused command-shape, invalid-lifetime, failure-precedence, replacement-owner, and force-release coverage. --- src/horizon/src/Lock.php | 32 +++++--- tests/Horizon/Unit/LockTest.php | 82 +++++++++++++++++++ .../Integration/Horizon/Feature/LockTest.php | 34 ++++++++ 3 files changed, 135 insertions(+), 13 deletions(-) create mode 100644 tests/Horizon/Unit/LockTest.php create mode 100644 tests/Integration/Horizon/Feature/LockTest.php diff --git a/src/horizon/src/Lock.php b/src/horizon/src/Lock.php index 261e95022..e5c831e16 100644 --- a/src/horizon/src/Lock.php +++ b/src/horizon/src/Lock.php @@ -5,8 +5,10 @@ namespace Hypervel\Horizon; use Closure; +use Hypervel\Cache\RedisLock; use Hypervel\Contracts\Redis\Factory as Redis; use Hypervel\Redis\RedisProxy; +use InvalidArgumentException; class Lock { @@ -25,13 +27,9 @@ public function __construct( */ public function with(string $key, Closure $callback, int $seconds = 60): void { - if ($this->get($key, $seconds)) { - try { - call_user_func($callback); - } finally { - $this->release($key); - } - } + $this->assertPositiveLifetime($key, $seconds); + + (new RedisLock($this->connection(), $key, $seconds))->get($callback); } /** @@ -47,13 +45,9 @@ public function exists(string $key): bool */ public function get(string $key, int $seconds = 60): bool { - $result = $this->connection()->setNx($key, '1') === 1; + $this->assertPositiveLifetime($key, $seconds); - if ($result) { - $this->connection()->expire($key, $seconds); - } - - return $result; + return $this->connection()->set($key, '1', 'EX', $seconds, 'NX') === true; } /** @@ -64,6 +58,18 @@ public function release(string $key): void $this->connection()->del($key); } + /** + * Ensure the lock lifetime is positive. + */ + private function assertPositiveLifetime(string $key, int $seconds): void + { + if ($seconds <= 0) { + throw new InvalidArgumentException( + "Horizon lock [{$key}] requires a positive lifetime; {$seconds} given." + ); + } + } + /** * Get the Redis connection instance. */ diff --git a/tests/Horizon/Unit/LockTest.php b/tests/Horizon/Unit/LockTest.php new file mode 100644 index 000000000..0f01da9b1 --- /dev/null +++ b/tests/Horizon/Unit/LockTest.php @@ -0,0 +1,82 @@ +shouldReceive('set') + ->once() + ->with('metrics', '1', 'EX', 60, 'NX') + ->andReturnTrue(); + $connection->shouldNotReceive('setNx'); + $connection->shouldNotReceive('expire'); + + $lock = new Lock($this->redisFactory($connection)); + + $this->assertTrue($lock->get('metrics')); + } + + public function testGetRejectsNonPositiveLifetimesBeforeConnecting(): void + { + $redis = m::mock(Factory::class); + $redis->shouldNotReceive('connection'); + + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage('Horizon lock [metrics] requires a positive lifetime; 0 given.'); + + (new Lock($redis))->get('metrics', 0); + } + + public function testWithRejectsNonPositiveLifetimesBeforeConnecting(): void + { + $redis = m::mock(Factory::class); + $redis->shouldNotReceive('connection'); + + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage('Horizon lock [metrics] requires a positive lifetime; -1 given.'); + + (new Lock($redis))->with('metrics', static function (): void { + }, -1); + } + + public function testCallbackFailureRemainsPrimaryWhenOwnedReleaseFails(): void + { + $callbackFailure = new RuntimeException('callback failed'); + $rawConnection = m::mock(RedisConnection::class); + $rawConnection->shouldReceive('pack')->once()->andReturn(['packed-owner']); + $rawConnection->shouldReceive('eval')->once()->andThrow(new RuntimeException('release failed')); + + $connection = m::mock(RedisProxy::class); + $connection->shouldReceive('set') + ->once() + ->with('metrics', m::type('string'), 'EX', 60, 'NX') + ->andReturnTrue(); + $connection->shouldReceive('withConnection') + ->once() + ->andReturnUsing(static fn (callable $callback): mixed => $callback($rawConnection)); + + try { + (new Lock($this->redisFactory($connection)))->with( + 'metrics', + static fn () => throw $callbackFailure, + ); + $this->fail('Expected the callback failure to propagate.'); + } catch (RuntimeException $thrown) { + $this->assertSame($callbackFailure, $thrown); + } + } +} diff --git a/tests/Integration/Horizon/Feature/LockTest.php b/tests/Integration/Horizon/Feature/LockTest.php new file mode 100644 index 000000000..b1e0ada22 --- /dev/null +++ b/tests/Integration/Horizon/Feature/LockTest.php @@ -0,0 +1,34 @@ +app->make(Factory::class)->connection('horizon'); + $lock = $this->app->make(Lock::class); + + $lock->with('owned-lock', static function () use ($connection): void { + $connection->set('owned-lock', 'replacement-owner', 'EX', 60); + }); + + $this->assertSame('replacement-owner', $connection->get('owned-lock')); + } + + public function testReleaseRemainsAnExplicitForceRelease(): void + { + $connection = $this->app->make(Factory::class)->connection('horizon'); + $connection->set('owned-lock', 'another-owner', 'EX', 60); + + $this->app->make(Lock::class)->release('owned-lock'); + + $this->assertNull($connection->get('owned-lock')); + } +} From 9d666ab2cc8641735db1676317f66be36b6349be Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:39:48 +0000 Subject: [PATCH 08/24] fix: make Horizon controller queries portable and exact Escape batch-search patterns with an explicit portable character so literal percent and underscore searches work on MySQL, MariaDB, and SQLite. Preserve case-insensitive matching and exact zero cursor behavior. Make failed-job tag filtering distinguish null and empty input from the valid string zero, reuse the resolved tag consistently, and expose truthful controller result types. Add unit, SQLite, and MySQL regressions. --- .../Http/Controllers/BatchesController.php | 10 +- .../Http/Controllers/FailedJobsController.php | 15 +-- .../Horizon/Unit/FailedJobsControllerTest.php | 49 ++++++++++ .../Controller/BatchesControllerTest.php | 17 ++++ .../Database/MySql/BatchesControllerTest.php | 91 +++++++++++++++++++ 5 files changed, 173 insertions(+), 9 deletions(-) create mode 100644 tests/Horizon/Unit/FailedJobsControllerTest.php create mode 100644 tests/Integration/Horizon/Database/MySql/BatchesControllerTest.php diff --git a/src/horizon/src/Http/Controllers/BatchesController.php b/src/horizon/src/Http/Controllers/BatchesController.php index 172f114dd..8eb328b95 100644 --- a/src/horizon/src/Http/Controllers/BatchesController.php +++ b/src/horizon/src/Http/Controllers/BatchesController.php @@ -48,15 +48,19 @@ public function index(Request $request): array */ private function searchBatches(Request $request): array { - $pattern = '%' . addcslashes($request->query('query'), '\%_') . '%'; + $pattern = '%' . str_replace( + ['!', '%', '_'], + ['!!', '!%', '!_'], + $request->query('query') + ) . '%'; $beforeId = $request->query('before_id'); return DB::connection(Config::get('queue.batching.database')) ->table(Config::string('queue.batching.table')) ->where(function ($q) use ($pattern) { - $q->whereRaw("lower(name) like lower(?) escape '\\'", [$pattern]) - ->orWhereRaw("lower(id) like lower(?) escape '\\'", [$pattern]); + $q->whereRaw("lower(name) like lower(?) escape '!'", [$pattern]) + ->orWhereRaw("lower(id) like lower(?) escape '!'", [$pattern]); }) ->orderByDesc('id') ->limit(50) diff --git a/src/horizon/src/Http/Controllers/FailedJobsController.php b/src/horizon/src/Http/Controllers/FailedJobsController.php index 2c2125b5a..5ab3d47f0 100644 --- a/src/horizon/src/Http/Controllers/FailedJobsController.php +++ b/src/horizon/src/Http/Controllers/FailedJobsController.php @@ -26,12 +26,15 @@ public function __construct( */ public function index(Request $request): array { - $jobs = ! $request->query('tag') + $tag = $request->query('tag'); + $hasTag = $tag !== null && $tag !== ''; + + $jobs = ! $hasTag ? $this->paginate($request) - : $this->paginateByTag($request, $request->query('tag')); + : $this->paginateByTag($request, $tag); - $total = $request->query('tag') - ? $this->tags->count('failed:' . $request->query('tag')) + $total = $hasTag + ? $this->tags->count('failed:' . $tag) : $this->jobs->countFailed(); return [ @@ -43,7 +46,7 @@ public function index(Request $request): array /** * Paginate the failed jobs for the request. */ - protected function paginate(Request $request) + protected function paginate(Request $request): Collection { $startingAt = $request->query('starting_at') ?: -1; @@ -73,7 +76,7 @@ protected function paginateByTag(Request $request, string $tag): Collection /** * Get a failed job instance. */ - public function show(string $id): mixed + public function show(string $id): array { return (array) $this->jobs->getJobs([$id])->map(function ($job) { return $this->decode($job); diff --git a/tests/Horizon/Unit/FailedJobsControllerTest.php b/tests/Horizon/Unit/FailedJobsControllerTest.php new file mode 100644 index 000000000..f80ae6e4b --- /dev/null +++ b/tests/Horizon/Unit/FailedJobsControllerTest.php @@ -0,0 +1,49 @@ +shouldReceive('getJobs')->once()->with(['job-id'], 0)->andReturn(collect()); + $jobs->shouldReceive('countFailed')->never(); + + $tags = m::mock(TagRepository::class); + $tags->shouldReceive('paginate')->once()->with('failed:0', 0, 50)->andReturn(['job-id']); + $tags->shouldReceive('count')->once()->with('failed:0')->andReturn(1); + + $result = (new FailedJobsController($jobs, $tags))->index( + Request::create('/?tag=0'), + ); + + $this->assertSame(1, $result['total']); + } + + public function testEmptyTagUsesTheUnfilteredFailedJobList(): void + { + $jobs = m::mock(JobRepository::class); + $jobs->shouldReceive('getFailed')->once()->with(-1)->andReturn(collect()); + $jobs->shouldReceive('countFailed')->once()->andReturn(2); + + $tags = m::mock(TagRepository::class); + $tags->shouldReceive('paginate')->never(); + $tags->shouldReceive('count')->never(); + + $result = (new FailedJobsController($jobs, $tags))->index( + Request::create('/?tag='), + ); + + $this->assertSame(2, $result['total']); + } +} diff --git a/tests/Integration/Horizon/Controller/BatchesControllerTest.php b/tests/Integration/Horizon/Controller/BatchesControllerTest.php index 4a023d565..3c38356ee 100644 --- a/tests/Integration/Horizon/Controller/BatchesControllerTest.php +++ b/tests/Integration/Horizon/Controller/BatchesControllerTest.php @@ -71,6 +71,23 @@ public function testSearchEscapesLikeWildcards() $this->assertEmpty($response->original['batches']); } + public function testSearchMatchesLiteralUnderscores(): void + { + $this->setupBatchTable(); + $this->seedBatches(); + $this->insertBatch('batch_under_score', 'Import_Users'); + + $response = $this->actingAs(new Fakes\User) + ->get('/horizon/api/batches?query=_'); + + $response->assertOk(); + + $batches = $response->original['batches']; + + $this->assertCount(1, $batches); + $this->assertSame('batch_under_score', $batches[0]->id); + } + public function testSearchSupportsCursorPagination() { $this->setupBatchTable(); diff --git a/tests/Integration/Horizon/Database/MySql/BatchesControllerTest.php b/tests/Integration/Horizon/Database/MySql/BatchesControllerTest.php new file mode 100644 index 000000000..3629b8bed --- /dev/null +++ b/tests/Integration/Horizon/Database/MySql/BatchesControllerTest.php @@ -0,0 +1,91 @@ +make('config')->set('queue.batching', [ + 'database' => $app->make('config')->string('database.default'), + 'table' => 'job_batches', + ]); + } + + protected function afterRefreshingDatabase(): void + { + Schema::create('job_batches', static function (Blueprint $table): void { + $table->string('id')->primary(); + $table->string('name'); + $table->integer('total_jobs'); + $table->integer('pending_jobs'); + $table->integer('failed_jobs'); + $table->longText('failed_job_ids'); + $table->mediumText('options')->nullable(); + $table->integer('cancelled_at')->nullable(); + $table->integer('created_at'); + $table->integer('finished_at')->nullable(); + }); + + $this->insertBatch('plain', 'Import Users'); + $this->insertBatch('percent', 'Import%Users'); + $this->insertBatch('underscore', 'Import_Users'); + $this->insertBatch('wildcard-decoy', 'ImportXUsers'); + } + + protected function destroyDatabaseMigrations(): void + { + Schema::dropIfExists('job_batches'); + } + + public function testBatchSearchUsesPortableLiteralWildcardEscaping(): void + { + $controller = new BatchesController($this->app->make(BatchRepository::class)); + + $plain = $controller->index(Request::create('/?query=Import Users')); + $percent = $controller->index(Request::create('/?query=%25')); + $underscore = $controller->index(Request::create('/?query=_')); + + $this->assertSame(['plain'], array_column($plain['batches'], 'id')); + $this->assertSame(['percent'], array_column($percent['batches'], 'id')); + $this->assertSame(['underscore'], array_column($underscore['batches'], 'id')); + } + + private function insertBatch(string $id, string $name): void + { + DB::table('job_batches')->insert([ + 'id' => $id, + 'name' => $name, + 'total_jobs' => 0, + 'pending_jobs' => 0, + 'failed_jobs' => 0, + 'failed_job_ids' => '[]', + 'options' => serialize([]), + 'created_at' => time(), + 'cancelled_at' => null, + 'finished_at' => null, + ]); + } +} From 168fcef9de4f9a8b89538ad4c52dd105f31aeead Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:39:54 +0000 Subject: [PATCH 09/24] fix: restore the Horizon worker command contract Restore the inherited stop-when-empty-for option that Horizon omitted while replacing the Queue worker signature. Accept the parent command nullable result instead of converting valid maintenance and once-mode paths into type errors. Exercise the real command with a captured WorkerOptions instance and compare its option surface with the base Queue command, allowing only the Horizon supervisor addition. --- src/horizon/src/Console/WorkCommand.php | 5 +- tests/Horizon/Console/WorkCommandTest.php | 63 +++++++++++++++++++++++ 2 files changed, 66 insertions(+), 2 deletions(-) create mode 100644 tests/Horizon/Console/WorkCommandTest.php diff --git a/src/horizon/src/Console/WorkCommand.php b/src/horizon/src/Console/WorkCommand.php index e88bd06e6..27925e42a 100644 --- a/src/horizon/src/Console/WorkCommand.php +++ b/src/horizon/src/Console/WorkCommand.php @@ -19,6 +19,7 @@ class WorkCommand extends BaseWorkCommand {--once : Only process the next job on the queue} {--concurrency=1 : The number of jobs to process at once} {--stop-when-empty : Stop when the queue is empty} + {--stop-when-empty-for=0 : Stop when the queue has been empty for the given number of seconds} {--delay=0 : The number of seconds to delay failed jobs (Deprecated)} {--backoff=0 : The number of seconds to wait before retrying a job that encountered an uncaught exception} {--max-jobs=0 : The number of jobs to process before stopping} @@ -41,9 +42,9 @@ class WorkCommand extends BaseWorkCommand /** * Execute the console command. */ - public function handle(): int + public function handle(): ?int { - if (config('horizon.fast_termination')) { + if (config()->boolean('horizon.fast_termination')) { ignore_user_abort(true); } diff --git a/tests/Horizon/Console/WorkCommandTest.php b/tests/Horizon/Console/WorkCommandTest.php new file mode 100644 index 000000000..4ad307875 --- /dev/null +++ b/tests/Horizon/Console/WorkCommandTest.php @@ -0,0 +1,63 @@ +shouldReceive('setName') + ->once() + ->with('default') + ->andReturnSelf(); + $worker->shouldReceive('setCache') + ->once() + ->andReturnSelf(); + $worker->shouldReceive('runNextJob') + ->once() + ->withArgs(function (string $connection, string $queue, WorkerOptions $options): bool { + $this->assertSame('sync', $connection); + $this->assertSame('default', $queue); + $this->assertSame(7, $options->stopWhenEmptyFor); + + return true; + }) + ->andReturnNull(); + + $this->app->instance(Worker::class, $worker); + + $this->artisan('horizon:work', [ + 'connection' => 'sync', + '--once' => true, + '--stop-when-empty-for' => 7, + ])->assertSuccessful(); + } + + public function testHorizonWorkKeepsTheQueueWorkerOptionSurface(): void + { + $artisan = $this->app->make(Kernel::class)->getArtisan(); + $queueOptions = array_keys($artisan->find('queue:work')->getDefinition()->getOptions()); + $horizonOptions = array_keys($artisan->find('horizon:work')->getDefinition()->getOptions()); + + $this->assertSame([], array_values(array_diff($queueOptions, $horizonOptions))); + $this->assertSame(['supervisor'], array_values(array_diff($horizonOptions, $queueOptions))); + } +} From ab2bbc2222c3a97d7c616c2d7f2c90b8cca08628 Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:40:10 +0000 Subject: [PATCH 10/24] fix: preserve Horizon process termination state and exit codes Replace public loop flags with protected nullable terminal status, clear pending signals on termination, and stop later commands or monitor work from reopening a terminated owner. Return the exact status through master, supervisor, and console command boundaries. Preserve intended restart and terminal exit-code classes, including nullable process results and literal zero names. Add isolated signal and command regressions plus a persisted-master readiness barrier so loaded test workers cannot signal before the handler exists or hang after barrier failure. --- src/horizon/src/Console/HorizonCommand.php | 9 +-- src/horizon/src/Console/SupervisorCommand.php | 12 ++-- src/horizon/src/Console/TerminateCommand.php | 8 +-- .../MonitorMasterSupervisorMemory.php | 2 +- src/horizon/src/MasterSupervisor.php | 35 ++++++--- src/horizon/src/ProcessPool.php | 2 +- src/horizon/src/Supervisor.php | 26 +++++-- src/horizon/src/SupervisorProcess.php | 2 +- .../Fakes/SupervisorWithFakeMonitor.php | 8 ++- .../Fixtures/FakeSupervisorFactory.php | 9 ++- .../Horizon/Feature/HorizonCommandTest.php | 51 +++++++++++++ .../Horizon/Feature/MasterSupervisorTest.php | 50 +++++++------ .../Horizon/Feature/ProcessExitStatusTest.php | 71 +++++++++++++++++++ .../Horizon/Feature/SupervisorCommandTest.php | 24 ++++++- .../Horizon/Feature/SupervisorTest.php | 4 +- 15 files changed, 241 insertions(+), 72 deletions(-) create mode 100644 tests/Integration/Horizon/Feature/HorizonCommandTest.php create mode 100644 tests/Integration/Horizon/Feature/ProcessExitStatusTest.php diff --git a/src/horizon/src/Console/HorizonCommand.php b/src/horizon/src/Console/HorizonCommand.php index 6d69eaa68..4e029a573 100644 --- a/src/horizon/src/Console/HorizonCommand.php +++ b/src/horizon/src/Console/HorizonCommand.php @@ -26,11 +26,12 @@ class HorizonCommand extends Command /** * Execute the console command. */ - public function handle(MasterSupervisorRepository $masters): void + public function handle(MasterSupervisorRepository $masters): int { if ($masters->find(MasterSupervisor::name())) { $this->components->warn('A master supervisor is already running on this machine.'); - return; + + return self::SUCCESS; } $environment = $this->option('environment') ?? config('horizon.env') ?? config('app.env'); @@ -50,9 +51,9 @@ public function handle(MasterSupervisorRepository $masters): void $this->components->info('Shutting down.'); - return $master->terminate(); // @phpstan-ignore method.void + $master->terminate(); }); - $master->monitor(); + return $master->monitor(); } } diff --git a/src/horizon/src/Console/SupervisorCommand.php b/src/horizon/src/Console/SupervisorCommand.php index e1fa18c5c..d6bad0c12 100644 --- a/src/horizon/src/Console/SupervisorCommand.php +++ b/src/horizon/src/Console/SupervisorCommand.php @@ -55,7 +55,7 @@ class SupervisorCommand extends Command /** * Execute the console command. */ - public function handle(SupervisorFactory $factory): ?int + public function handle(SupervisorFactory $factory): int { $supervisor = $factory->make( $this->supervisorOptions() @@ -69,15 +69,13 @@ public function handle(SupervisorFactory $factory): ?int return 13; } - $this->start($supervisor); - - return 0; + return $this->start($supervisor); } /** * Start the given supervisor. */ - protected function start(Supervisor $supervisor): void + protected function start(Supervisor $supervisor): int { if ($supervisor->options->nice) { proc_nice($supervisor->options->nice); @@ -97,7 +95,7 @@ protected function start(Supervisor $supervisor): void $balancedWorkerCount - $supervisor->totalSystemProcessCount() )); - $supervisor->monitor(); + return $supervisor->monitor(); } /** @@ -105,7 +103,7 @@ protected function start(Supervisor $supervisor): void */ protected function supervisorOptions(): SupervisorOptions { - $balance = $this->option('balance'); + $balance = $this->option('balance') ?? 'off'; $autoScalingStrategy = $balance === 'auto' ? $this->option('auto-scaling-strategy') : null; diff --git a/src/horizon/src/Console/TerminateCommand.php b/src/horizon/src/Console/TerminateCommand.php index 386a98655..2023bf3e4 100644 --- a/src/horizon/src/Console/TerminateCommand.php +++ b/src/horizon/src/Console/TerminateCommand.php @@ -35,9 +35,8 @@ class TerminateCommand extends Command */ public function handle(CacheFactory $cache, MasterSupervisorRepository $masters): void { - if (config('horizon.fast_termination')) { - /* @phpstan-ignore-next-line */ - $cache->forever( + if (config()->boolean('horizon.fast_termination')) { + $cache->store()->forever( 'horizon:terminate:wait', $this->option('wait') ); @@ -62,7 +61,6 @@ public function handle(CacheFactory $cache, MasterSupervisorRepository $masters) } })->whenNotEmpty(fn () => $this->output->writeln('')); - $this->hypervel->make(CacheFactory::class) - ->store()->forever(Worker::RESTART_SIGNAL_CACHE_KEY, $this->currentTime()); + $cache->store()->forever(Worker::RESTART_SIGNAL_CACHE_KEY, $this->currentTime()); } } diff --git a/src/horizon/src/Listeners/MonitorMasterSupervisorMemory.php b/src/horizon/src/Listeners/MonitorMasterSupervisorMemory.php index d7cbc8e87..9b3144454 100644 --- a/src/horizon/src/Listeners/MonitorMasterSupervisorMemory.php +++ b/src/horizon/src/Listeners/MonitorMasterSupervisorMemory.php @@ -21,7 +21,7 @@ public function handle(MasterSupervisorLooped $event): void $master = $event->master; - $memoryLimit = config('horizon.memory_limit', 64); + $memoryLimit = config()->integer('horizon.memory_limit'); if ($master->memoryUsage() > $memoryLimit) { event(new MasterSupervisorOutOfMemory($master)); diff --git a/src/horizon/src/MasterSupervisor.php b/src/horizon/src/MasterSupervisor.php index 5cba8324d..5f67803ea 100644 --- a/src/horizon/src/MasterSupervisor.php +++ b/src/horizon/src/MasterSupervisor.php @@ -56,7 +56,10 @@ class MasterSupervisor implements Pausable, Restartable, Terminable */ protected static ?string $token = null; - public bool $shouldExitLoop = false; + /** + * The terminal exit status for this master supervisor. + */ + protected ?int $exitStatus = null; /** * Create a new master supervisor instance. @@ -176,18 +179,18 @@ public function terminate(int $status = 0): void sleep(1); } - if (config('horizon.fast_termination')) { - /* @phpstan-ignore-next-line */ - app(CacheFactory::class)->forget('horizon:terminate:wait'); + if (config()->boolean('horizon.fast_termination')) { + app(CacheFactory::class)->store()->forget('horizon:terminate:wait'); } - $this->shouldExitLoop = true; + $this->pendingSignals = []; + $this->exitStatus = $status; } /** * Monitor the worker processes. */ - public function monitor(): void + public function monitor(): int { $this->ensureNoOtherMasterSupervisors(); @@ -198,11 +201,11 @@ public function monitor(): void while (true) { sleep(1); - if ($this->shouldExitLoop) { - break; - } - $this->loop(); + + if ($this->exitStatus !== null) { + return $this->exitStatus; + } } } @@ -228,6 +231,10 @@ public function loop(): void $this->processPendingCommands(); + if ($this->exitStatus !== null) { + return; + } + if ($this->working) { $this->monitorSupervisors(); } @@ -246,6 +253,10 @@ public function loop(): void protected function processPendingCommands(): void { foreach (app(HorizonCommandQueue::class)->pending($this->commandQueue()) as $command) { + if ($this->exitStatus !== null) { + return; + } + app($command->command)->process($this, $command->options); } } @@ -299,7 +310,9 @@ public static function commandQueue(): string */ public static function commandQueueFor(?string $name = null): string { - return $name ? 'master:' . $name : static::commandQueue(); + return $name === null || $name === '' + ? static::commandQueue() + : 'master:' . $name; } /** diff --git a/src/horizon/src/ProcessPool.php b/src/horizon/src/ProcessPool.php index 192bdc7cb..152d27838 100644 --- a/src/horizon/src/ProcessPool.php +++ b/src/horizon/src/ProcessPool.php @@ -145,7 +145,7 @@ protected function start(): static */ protected function createProcess(): WorkerProcess { - $class = config('horizon.fast_termination') + $class = config()->boolean('horizon.fast_termination') ? BackgroundProcess::class : Process::class; diff --git a/src/horizon/src/Supervisor.php b/src/horizon/src/Supervisor.php index f980add57..1abb7b271 100644 --- a/src/horizon/src/Supervisor.php +++ b/src/horizon/src/Supervisor.php @@ -49,7 +49,10 @@ class Supervisor implements Pausable, Restartable, Terminable */ public ?Closure $output = null; - public bool $shouldExitLoop = false; + /** + * The terminal exit status for this supervisor. + */ + protected ?int $exitStatus = null; /** * Create a new supervisor instance. @@ -188,7 +191,8 @@ public function terminate(int $status = 0): void } } - $this->shouldExitLoop = true; + $this->pendingSignals = []; + $this->exitStatus = $status; } /** @@ -196,14 +200,14 @@ public function terminate(int $status = 0): void */ protected function shouldWait(): bool { - // @phpstan-ignore-next-line - return ! config('horizon.fast_termination') || app(CacheFactory::class)->get('horizon:terminate:wait'); + return ! config()->boolean('horizon.fast_termination') + || app(CacheFactory::class)->store()->get('horizon:terminate:wait'); } /** * Monitor the worker processes. */ - public function monitor(): void + public function monitor(): int { $this->ensureNoDuplicateSupervisors(); @@ -216,8 +220,8 @@ public function monitor(): void $this->loop(); - if ($this->shouldExitLoop) { - break; + if ($this->exitStatus !== null) { + return $this->exitStatus; } } } @@ -246,6 +250,10 @@ public function loop(): void $this->processPendingCommands(); + if ($this->exitStatus !== null) { + return; + } + // If the supervisor is working, we will perform any needed scaling operations and // monitor all of these underlying worker processes to make sure they are still // processing queued jobs. If they have died, we will restart them each here. @@ -280,6 +288,10 @@ protected function ensureParentIsRunning(): void protected function processPendingCommands(): void { foreach (app(HorizonCommandQueue::class)->pending($this->name) as $command) { + if ($this->exitStatus !== null) { + return; + } + app($command->command)->process($this, $command->options); } } diff --git a/src/horizon/src/SupervisorProcess.php b/src/horizon/src/SupervisorProcess.php index d8cce64c2..850afc5a7 100644 --- a/src/horizon/src/SupervisorProcess.php +++ b/src/horizon/src/SupervisorProcess.php @@ -98,7 +98,7 @@ public function monitor(): void // If the supervisor exited with a status code that we do not restart on then // we will not attempt to restart it. Otherwise, we will need to provision // it back out based on the latest provisioning information we have now. - if (in_array($exitCode, $this->dontRestartOn)) { + if (in_array($exitCode, $this->dontRestartOn, true)) { return; } diff --git a/tests/Integration/Horizon/Feature/Fakes/SupervisorWithFakeMonitor.php b/tests/Integration/Horizon/Feature/Fakes/SupervisorWithFakeMonitor.php index 6834de427..b67ca34a7 100644 --- a/tests/Integration/Horizon/Feature/Fakes/SupervisorWithFakeMonitor.php +++ b/tests/Integration/Horizon/Feature/Fakes/SupervisorWithFakeMonitor.php @@ -8,10 +8,14 @@ class SupervisorWithFakeMonitor extends Supervisor { - public $monitoring = false; + public bool $monitoring = false; - public function monitor(): void + public int $monitorStatus = 0; + + public function monitor(): int { $this->monitoring = true; + + return $this->monitorStatus; } } diff --git a/tests/Integration/Horizon/Feature/Fixtures/FakeSupervisorFactory.php b/tests/Integration/Horizon/Feature/Fixtures/FakeSupervisorFactory.php index 0a9f64b12..887205346 100644 --- a/tests/Integration/Horizon/Feature/Fixtures/FakeSupervisorFactory.php +++ b/tests/Integration/Horizon/Feature/Fixtures/FakeSupervisorFactory.php @@ -11,10 +11,15 @@ class FakeSupervisorFactory extends SupervisorFactory { - public $supervisor; + public ?SupervisorWithFakeMonitor $supervisor = null; + + public int $monitorStatus = 0; public function make(SupervisorOptions $options): Supervisor { - return $this->supervisor = new SupervisorWithFakeMonitor($options); + $this->supervisor = new SupervisorWithFakeMonitor($options); + $this->supervisor->monitorStatus = $this->monitorStatus; + + return $this->supervisor; } } diff --git a/tests/Integration/Horizon/Feature/HorizonCommandTest.php b/tests/Integration/Horizon/Feature/HorizonCommandTest.php new file mode 100644 index 000000000..9a6b29bb9 --- /dev/null +++ b/tests/Integration/Horizon/Feature/HorizonCommandTest.php @@ -0,0 +1,51 @@ +persist(); + + $this->artisan('horizon') + ->expectsOutputToContain('A master supervisor is already running on this machine.') + ->assertExitCode(0); + } + + #[RunInSeparateProcess] + public function testCommandReturnsTheMasterMonitorStatus(): void + { + $barrierReached = false; + + Coroutine::create(function () use (&$barrierReached): void { + try { + $masters = app(MasterSupervisorRepository::class); + $deadline = hrtime(true) + 10_000_000_000; + + while (hrtime(true) < $deadline) { + if ($masters->find(MasterSupervisor::name()) !== null) { + $barrierReached = true; + break; + } + + Coroutine::sleep(0.01); + } + } finally { + // Always release the foreground command so a failed barrier cannot hang the test. + posix_kill(getmypid(), SIGINT); + } + }); + + $this->artisan('horizon')->assertExitCode(0); + $this->assertTrue($barrierReached, 'Horizon master never registered before the SIGINT barrier.'); + } +} diff --git a/tests/Integration/Horizon/Feature/MasterSupervisorTest.php b/tests/Integration/Horizon/Feature/MasterSupervisorTest.php index 9bc022ebc..6eb9c2668 100644 --- a/tests/Integration/Horizon/Feature/MasterSupervisorTest.php +++ b/tests/Integration/Horizon/Feature/MasterSupervisorTest.php @@ -20,6 +20,7 @@ use Hypervel\Tests\Integration\Horizon\Feature\Fixtures\SupervisorProcessWithFakeRestart; use Hypervel\Tests\Integration\Horizon\IntegrationTestCase; use Mockery as m; +use PHPUnit\Framework\Attributes\DataProvider; use Symfony\Component\Process\Process; class MasterSupervisorTest extends IntegrationTestCase @@ -37,7 +38,8 @@ public function testNamesCanBeCustomized() $this->assertStringStartsWith('test-name', $master->name()); } - public function testMasterProcessMarksCleanExitsAsDeadAndRemovesThem() + #[DataProvider('terminalExitCodes')] + public function testMasterProcessMarksTerminalExitsAsDeadAndRemovesThem(int $exitCode): void { $process = m::mock(Process::class); $master = new MasterSupervisor; @@ -49,7 +51,7 @@ public function testMasterProcessMarksCleanExitsAsDeadAndRemovesThem() $process->shouldReceive('isStarted')->andReturn(true); $process->shouldReceive('isRunning')->andReturn(false); - $process->shouldReceive('getExitCode')->andReturn(0); + $process->shouldReceive('getExitCode')->andReturn($exitCode); $master->loop(); @@ -57,27 +59,14 @@ public function testMasterProcessMarksCleanExitsAsDeadAndRemovesThem() $this->assertCount(0, $master->supervisors); } - public function testMasterProcessMarksDuplicatesAsDeadAndRemovesThem() + public static function terminalExitCodes(): array { - $process = m::mock(Process::class); - $master = new MasterSupervisor; - $master->working = true; - $master->supervisors[] = $supervisorProcess = new SupervisorProcess( - $this->supervisorOptions(), - $process - ); - - $process->shouldReceive('isStarted')->andReturn(true); - $process->shouldReceive('isRunning')->andReturn(false); - $process->shouldReceive('getExitCode')->andReturn(13); - - $master->loop(); - - $this->assertTrue($supervisorProcess->dead); - $this->assertCount(0, $master->supervisors); + // Exit 13 is caught by the duplicate-supervisor branch before dontRestartOn. + return [[0], [2], [13]]; } - public function testMasterProcessRestartsUnexpectedExits() + #[DataProvider('restartableExitCodes')] + public function testMasterProcessRestartsUnexpectedExits(?int $exitCode): void { $process = m::mock(Process::class); $master = new MasterSupervisor; @@ -89,7 +78,7 @@ public function testMasterProcessRestartsUnexpectedExits() $process->shouldReceive('isStarted')->andReturn(true); $process->shouldReceive('isRunning')->andReturn(false); - $process->shouldReceive('getExitCode')->andReturn(50); + $process->shouldReceive('getExitCode')->andReturn($exitCode); $master->loop(); @@ -108,6 +97,11 @@ public function testMasterProcessRestartsUnexpectedExits() $this->assertSame('default', $command->options['queue']); } + public static function restartableExitCodes(): array + { + return [[null], [1], [12], [50]]; + } + public function testMasterProcessRestartsProcessesThatNeverStarted() { $process = m::mock(Process::class); @@ -228,7 +222,7 @@ public function testSupervisorRepositoryReturnsNullIfNoSupervisorExistsWithGiven $this->assertNull($repository->find('nothing')); } - public function testSupervisorProcessTerminatesAllWorkersAndExitsOnFullTermination() + public function testSupervisorProcessTerminatesAllWorkersAndExitsOnFullTermination(): void { $master = new MasterSupervisor; $master->working = true; @@ -236,13 +230,11 @@ public function testSupervisorProcessTerminatesAllWorkersAndExitsOnFullTerminati $master->persist(); $master->terminate(); - $this->assertTrue($master->shouldExitLoop); - // Assert that the supervisor is removed... $this->assertNull(resolve(MasterSupervisorRepository::class)->find($master->name)); } - public function testSupervisorContinuesTerminationIfSupervisorsTakeTooLong() + public function testSupervisorContinuesTerminationIfSupervisorsTakeTooLong(): void { $master = new MasterSupervisor; $master->working = true; @@ -252,10 +244,16 @@ public function testSupervisorContinuesTerminationIfSupervisorsTakeTooLong() $master->persist(); $master->terminate(); - $this->assertTrue($master->shouldExitLoop); $this->assertTrue($supervisor->killed); } + public function testLiteralZeroMasterNameIsPreservedInTheCommandQueue(): void + { + $this->assertSame('master:0', MasterSupervisor::commandQueueFor('0')); + $this->assertSame(MasterSupervisor::commandQueue(), MasterSupervisor::commandQueueFor('')); + $this->assertSame(MasterSupervisor::commandQueue(), MasterSupervisor::commandQueueFor()); + } + protected function supervisorOptions() { return tap(new SupervisorOptions(MasterSupervisor::name() . ':name', 'redis'), function ($options) { diff --git a/tests/Integration/Horizon/Feature/ProcessExitStatusTest.php b/tests/Integration/Horizon/Feature/ProcessExitStatusTest.php new file mode 100644 index 000000000..84fca01b6 --- /dev/null +++ b/tests/Integration/Horizon/Feature/ProcessExitStatusTest.php @@ -0,0 +1,71 @@ +push($supervisor->name, Terminate::class, ['status' => 12]); + $commands->push($supervisor->name, ContinueWorking::class); + $commands->push($supervisor->name, Scale::class, ['scale' => 2]); + + $this->assertSame(12, $supervisor->monitor()); + $this->assertFalse($supervisor->working); + $this->assertSame(0, $supervisor->totalProcessCount()); + $this->assertNull(app(SupervisorRepository::class)->find($supervisor->name)); + Event::assertNotDispatched(SupervisorLooped::class); + } + + #[RunInSeparateProcess] + public function testMasterTerminationReturnsItsStatusAndSkipsLaterCommands(): void + { + Event::fake([MasterSupervisorLooped::class]); + $master = new MasterSupervisorWithPendingTermination; + app(HorizonCommandQueue::class)->push( + $master->commandQueue(), + AddSupervisor::class, + (new SupervisorOptions('unwanted-supervisor', 'redis'))->toArray(), + ); + + $this->assertSame(12, $master->monitor()); + $this->assertCount(0, $master->supervisors); + $this->assertNull(app(MasterSupervisorRepository::class)->find($master->name)); + Event::assertNotDispatched(MasterSupervisorLooped::class); + } +} + +class MasterSupervisorWithPendingTermination extends MasterSupervisor +{ + protected function listenForSignals(): void + { + $this->pendingSignals['terminateWithStatus'] = 'terminateWithStatus'; + } + + protected function terminateWithStatus(): void + { + $this->terminate(12); + } +} diff --git a/tests/Integration/Horizon/Feature/SupervisorCommandTest.php b/tests/Integration/Horizon/Feature/SupervisorCommandTest.php index 9854eee52..73b8b27fd 100644 --- a/tests/Integration/Horizon/Feature/SupervisorCommandTest.php +++ b/tests/Integration/Horizon/Feature/SupervisorCommandTest.php @@ -43,15 +43,24 @@ public function setUp(): void ->registerCommand($this->app->make(SupervisorCommand::class)); } - public function testSupervisorCommandCanStartSupervisorMonitoring() + public function testSupervisorCommandCanStartSupervisorMonitoring(): void { $this->app->instance(SupervisorFactory::class, $factory = new FakeSupervisorFactory); - $this->artisan('horizon:supervisor', static::OPTIONS); + $this->artisan('horizon:supervisor', static::OPTIONS)->assertExitCode(0); $this->assertTrue($factory->supervisor->monitoring); $this->assertTrue($factory->supervisor->working); } + public function testSupervisorCommandPropagatesTheMonitorStatus(): void + { + $factory = new FakeSupervisorFactory; + $factory->monitorStatus = 12; + $this->app->instance(SupervisorFactory::class, $factory); + + $this->artisan('horizon:supervisor', static::OPTIONS)->assertExitCode(12); + } + public function testSupervisorCommandCanStartPausedSupervisors() { $this->app->instance(SupervisorFactory::class, $factory = new FakeSupervisorFactory); @@ -84,6 +93,17 @@ public function testSupervisorCommandDefaultsEmptyQueue(): void $this->assertSame('default', $factory->supervisor->options->queue); } + public function testSupervisorCommandDefaultsMissingBalanceToOff(): void + { + $this->app->instance(SupervisorFactory::class, $factory = new FakeSupervisorFactory); + $options = static::OPTIONS; + unset($options['--balance']); + + $this->artisan('horizon:supervisor', $options); + + $this->assertSame('off', $factory->supervisor->options->balance); + } + private function myNiceness() { $pid = getmypid(); diff --git a/tests/Integration/Horizon/Feature/SupervisorTest.php b/tests/Integration/Horizon/Feature/SupervisorTest.php index 8de39afa5..7b4732733 100644 --- a/tests/Integration/Horizon/Feature/SupervisorTest.php +++ b/tests/Integration/Horizon/Feature/SupervisorTest.php @@ -397,7 +397,7 @@ public function testTerminatingProcessesThatAreStuckAreHardStopped() $this->assertFalse($process->isRunning()); } - public function testSupervisorProcessTerminatesAllWorkersAndExitsOnFullTermination() + public function testSupervisorProcessTerminatesAllWorkersAndExitsOnFullTermination(): void { $this->supervisor = $supervisor = new Supervisor($this->supervisorOptions()); @@ -407,8 +407,6 @@ public function testSupervisorProcessTerminatesAllWorkersAndExitsOnFullTerminati $supervisor->persist(); $supervisor->terminate(); - $this->assertTrue($supervisor->shouldExitLoop); - // Assert that the supervisor is removed... $this->assertNull(app(SupervisorRepository::class)->find($supervisor->name)); } From 6ffe4db7480867e2712db2a2700d67b07dfd71b8 Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:40:19 +0000 Subject: [PATCH 11/24] feat: add current Horizon APIs and configuration Add request-scoped CSP nonce rendering and development command registration while preserving Horizon public call shapes. Keep nonce state coroutine-local and retain the supported SMS routing surface for the first-party notification channel. Declare proxy and snapshot-lock configuration, correct nested fallback ownership and failed-tag retention, preserve exact zero values, and use strict balancing and scaling boundaries. Update the provider to use typed container resolution and record intentional deprecated or inapplicable omissions at future-sync points. --- src/horizon/config/horizon.php | 13 +++- src/horizon/src/AutoScaler.php | 2 +- src/horizon/src/Horizon.php | 53 +++++++++++-- src/horizon/src/HorizonServiceProvider.php | 24 ++++-- .../src/Listeners/StoreTagsForFailedJob.php | 2 +- src/horizon/src/SupervisorOptions.php | 2 +- tests/Horizon/HorizonConfigTest.php | 75 +++++++++++++++++++ tests/Horizon/HorizonTest.php | 73 ++++++++++++++++++ .../Horizon/Feature/SupervisorOptionsTest.php | 8 ++ 9 files changed, 231 insertions(+), 21 deletions(-) create mode 100644 tests/Horizon/HorizonTest.php diff --git a/src/horizon/config/horizon.php b/src/horizon/config/horizon.php index 8a95a2345..afe7d4a4b 100644 --- a/src/horizon/config/horizon.php +++ b/src/horizon/config/horizon.php @@ -37,13 +37,16 @@ |-------------------------------------------------------------------------- | | This is the URI path where Horizon will be accessible from. Feel free - | to change this path to anything you like. Note that the URI will not - | affect the paths of its internal API that aren't exposed to users. + | to change this path to anything you like. The proxy path prefixes it + | when Horizon is served from a subdirectory behind a reverse proxy, so + | the dashboard can still reach its own internal API. | */ 'path' => env('HORIZON_PATH', 'horizon'), + 'proxy_path' => '', + /* |-------------------------------------------------------------------------- | Horizon Redis Connection @@ -145,8 +148,9 @@ |-------------------------------------------------------------------------- | | Here you can configure how many snapshots should be kept to display in - | the metrics graph. This will get used in combination with Horizon's - | `horizon:snapshot` schedule to define how long to retain metrics. + | the metrics graph. This works with the `horizon:snapshot` schedule to + | define retention. The snapshot lock prevents overlapping + | `horizon:snapshot` runs and should match their interval in seconds. | */ @@ -155,6 +159,7 @@ 'job' => 24, 'queue' => 24, ], + 'snapshot_lock' => 300, ], /* diff --git a/src/horizon/src/AutoScaler.php b/src/horizon/src/AutoScaler.php index 9892aee1a..77d478925 100644 --- a/src/horizon/src/AutoScaler.php +++ b/src/horizon/src/AutoScaler.php @@ -104,7 +104,7 @@ protected function numberOfWorkersPerQueue(Supervisor $supervisor, Collection $q return [$queue => $numberOfProcesses *= $supervisor->options->maxProcesses]; } - if ($timeToClearAll == 0 + if ($timeToClearAll === 0.0 && $supervisor->options->autoScaling() ) { return [ diff --git a/src/horizon/src/Horizon.php b/src/horizon/src/Horizon.php index 6d3ce3625..acd2bce00 100644 --- a/src/horizon/src/Horizon.php +++ b/src/horizon/src/Horizon.php @@ -6,6 +6,8 @@ use Closure; use Exception; +use Hypervel\Context\CoroutineContext; +use Hypervel\Foundation\DevCommands; use Hypervel\Http\Request; use Hypervel\Redis\RedisConnection; use Hypervel\Support\HtmlString; @@ -47,6 +49,11 @@ class Horizon 'Metrics', 'Locks', 'Processes', ]; + /** + * The context key for Horizon's CSP nonce attribute. + */ + protected const CSP_NONCE_CONTEXT_KEY = '__horizon.csp_nonce'; + /** * Determine if the given request can access the Horizon dashboard. */ @@ -121,10 +128,12 @@ public static function css(): HtmlString throw new RuntimeException('Unable to load the Horizon dashboard CSS.'); } + $nonceAttribute = CoroutineContext::get(self::CSP_NONCE_CONTEXT_KEY, ''); + return new HtmlString(<<{$light} - - + + + {$app} HTML); } @@ -139,22 +148,26 @@ public static function js(): HtmlString $horizon = Js::from(static::scriptVariables()); + $nonceAttribute = CoroutineContext::get(self::CSP_NONCE_CONTEXT_KEY, ''); + return new HtmlString(<< + HTML); } + // REMOVED: Deprecated Horizon::night() theme mutator. + /** * Get the default JavaScript variables for Horizon. */ public static function scriptVariables(): array { return [ - 'path' => config('horizon.path'), - 'proxy_path' => config('horizon.proxy_path', ''), + 'path' => config()->string('horizon.path'), + 'proxy_path' => config()->string('horizon.proxy_path'), ]; } @@ -198,6 +211,34 @@ public static function routeSmsNotificationsTo(string $number): static return new static; } + /** + * Set the CSP nonce to use for style and script tags. + * + * Call this from request middleware so the nonce is isolated to the + * current request coroutine. + */ + public static function cspNonce(string $nonce): static + { + CoroutineContext::set( + self::CSP_NONCE_CONTEXT_KEY, + ' nonce="' . $nonce . '"', + ); + + return new static; + } + + /** + * Register the Horizon development commands. + * + * Boot-only. The registrations persist for the worker lifetime and affect + * every subsequent development command invocation. + */ + public static function registerDevCommands(): void + { + DevCommands::artisan('horizon', 'horizon'); + DevCommands::except('queue'); + } + /** * Flush all static state. */ diff --git a/src/horizon/src/HorizonServiceProvider.php b/src/horizon/src/HorizonServiceProvider.php index f8e23ac40..f1dd574d5 100644 --- a/src/horizon/src/HorizonServiceProvider.php +++ b/src/horizon/src/HorizonServiceProvider.php @@ -5,6 +5,7 @@ namespace Hypervel\Horizon; use Hypervel\Contracts\Events\Dispatcher; +use Hypervel\Contracts\Redis\Factory as RedisFactory; use Hypervel\Horizon\Connectors\RedisConnector; use Hypervel\Queue\QueueManager; use Hypervel\Support\Facades\Route; @@ -20,6 +21,7 @@ class HorizonServiceProvider extends ServiceProvider */ public function boot(): void { + // REMOVED: Laravel Sentinel middleware has no Hypervel integration. $this->normalizeConfig(); $this->registerEvents(); $this->registerRoutes(); @@ -35,7 +37,7 @@ protected function normalizeConfig(): void { $config = $this->app->make('config'); - if (! $config->get('horizon.name')) { + if (($name = $config->get('horizon.name')) === null || $name === '') { $config->set('horizon.name', $config->string('app.name')); } } @@ -59,11 +61,13 @@ protected function registerEvents(): void */ protected function registerRoutes(): void { + $config = $this->app->make('config'); + Route::group([ - 'domain' => config('horizon.domain', null), - 'prefix' => config('horizon.path'), + 'domain' => $config->get('horizon.domain'), + 'prefix' => $config->string('horizon.path'), 'namespace' => 'Hypervel\Horizon\Http\Controllers', - 'middleware' => config('horizon.middleware', ['web']), + 'middleware' => $config->array('horizon.middleware'), ], function () { $this->loadRoutesFrom(__DIR__ . '/../routes/web.php'); }); @@ -111,6 +115,7 @@ protected function registerCommands(): void Console\ListenCommand::class, Console\PauseCommand::class, Console\PauseSupervisorCommand::class, + // REMOVED: Deprecated horizon:publish; use horizon:install. Console\PurgeCommand::class, Console\SupervisorCommand::class, Console\SupervisorStatusCommand::class, @@ -141,6 +146,8 @@ public function register(): void $this->configure(); $this->registerServices(); $this->registerQueueConnectors(); + + Horizon::registerDevCommands(); } /** @@ -153,7 +160,7 @@ protected function configure(): void 'horizon' ); - Horizon::use(config('horizon.use', 'default')); + Horizon::use(config()->string('horizon.use')); } /** @@ -173,9 +180,10 @@ protected function registerQueueConnectors(): void { $this->callAfterResolving(QueueManager::class, function (QueueManager $manager) { $manager->addConnector('redis', function () { - return new RedisConnector( - $this->app['redis'] - ); + /** @var RedisFactory $redis */ + $redis = $this->app->make('redis'); + + return new RedisConnector($redis); }); }); } diff --git a/src/horizon/src/Listeners/StoreTagsForFailedJob.php b/src/horizon/src/Listeners/StoreTagsForFailedJob.php index 2702cb054..d5208899c 100644 --- a/src/horizon/src/Listeners/StoreTagsForFailedJob.php +++ b/src/horizon/src/Listeners/StoreTagsForFailedJob.php @@ -29,7 +29,7 @@ public function handle(JobFailed $event): void })->all(); $this->tags->addTemporary( - config('horizon.trim.failed', 2880), + config('horizon.trim.failed', 10080), $event->payload->id(), $tags ); diff --git a/src/horizon/src/SupervisorOptions.php b/src/horizon/src/SupervisorOptions.php index 9f915a118..c0f49f8dc 100644 --- a/src/horizon/src/SupervisorOptions.php +++ b/src/horizon/src/SupervisorOptions.php @@ -94,7 +94,7 @@ public function withQueue(string $queue): static */ public function balancing(): bool { - return in_array($this->balance, ['simple', 'auto']); + return in_array($this->balance, ['simple', 'auto'], true); } /** diff --git a/tests/Horizon/HorizonConfigTest.php b/tests/Horizon/HorizonConfigTest.php index cd6852289..e11467934 100644 --- a/tests/Horizon/HorizonConfigTest.php +++ b/tests/Horizon/HorizonConfigTest.php @@ -4,11 +4,67 @@ namespace Hypervel\Tests\Horizon; +use Hypervel\Config\Repository as ConfigRepository; +use Hypervel\Foundation\Application; +use Hypervel\Foundation\Configuration\ConfigMutationTracker; +use Hypervel\Horizon\HorizonServiceProvider; use Hypervel\Support\Env; +use Hypervel\Support\ServiceProvider; use Hypervel\Tests\TestCase; +use Mockery as m; class HorizonConfigTest extends TestCase { + public function testCanonicalDefaultsAreDeclared(): void + { + $config = require dirname(__DIR__, 2) . '/src/horizon/config/horizon.php'; + + $this->assertSame('', $config['proxy_path']); + $this->assertSame(300, $config['metrics']['snapshot_lock']); + $this->assertSame('horizon', $config['path']); + $this->assertSame('default', $config['use']); + $this->assertSame(['web'], $config['middleware']); + $this->assertFalse($config['fast_termination']); + $this->assertSame(64, $config['memory_limit']); + } + + public function testApplicationMetricsConfigurationReplacesPackageDefaults(): void + { + $config = new ConfigRepository([ + 'horizon' => [ + 'metrics' => [ + 'trim_snapshots' => ['job' => 12, 'queue' => 12], + ], + ], + ]); + $app = m::mock(Application::class)->makePartial(); + $app->shouldReceive('configurationIsCached')->andReturnFalse(); + $app->shouldReceive('make')->with('config')->andReturn($config); + $app->shouldReceive('make')->with(ConfigMutationTracker::class)->andReturn(new ConfigMutationTracker); + + (new HorizonConfigServiceProvider($app))->register(); + + $this->assertSame([ + 'trim_snapshots' => ['job' => 12, 'queue' => 12], + ], $config->get('horizon.metrics')); + } + + public function testOnlyMissingAndBlankNamesUseTheApplicationName(): void + { + foreach ([null, '', '0'] as $name) { + $config = new ConfigRepository([ + 'app' => ['name' => 'Hypervel'], + 'horizon' => ['name' => $name], + ]); + $app = m::mock(Application::class)->makePartial(); + $app->shouldReceive('make')->with('config')->andReturn($config); + + (new HorizonServiceProviderForTesting($app))->normalize(); + + $this->assertSame($name === '0' ? '0' : 'Hypervel', $config->get('horizon.name')); + } + } + public function testMissingAndBlankPrefixUseApplicationScopedDefault(): void { $key = 'HORIZON_PREFIX'; @@ -56,3 +112,22 @@ public function testMissingAndBlankPrefixUseApplicationScopedDefault(): void } } } + +class HorizonConfigServiceProvider extends ServiceProvider +{ + public function register(): void + { + $this->mergeConfigFrom( + dirname(__DIR__, 2) . '/src/horizon/config/horizon.php', + 'horizon', + ); + } +} + +class HorizonServiceProviderForTesting extends HorizonServiceProvider +{ + public function normalize(): void + { + $this->normalizeConfig(); + } +} diff --git a/tests/Horizon/HorizonTest.php b/tests/Horizon/HorizonTest.php new file mode 100644 index 000000000..262b74a99 --- /dev/null +++ b/tests/Horizon/HorizonTest.php @@ -0,0 +1,73 @@ +assertStringContainsString('