diff --git a/docs/legacy-overlap-release.md b/docs/legacy-overlap-release.md new file mode 100644 index 000000000..0258771a2 --- /dev/null +++ b/docs/legacy-overlap-release.md @@ -0,0 +1,32 @@ +# Legacy overlap release delay + +Applications using `Workflow\Workflow` can back off queue jobs that cannot +acquire an overlap lock: + +```php +use Workflow\WorkflowOptions; +use Workflow\WorkflowStub; + +$workflow = WorkflowStub::make(MyWorkflow::class); +$workflow->start('argument', new WorkflowOptions(overlapReleaseAfter: 5)); +``` + +`WorkflowOptions::set(['overlapReleaseAfter' => 5])` accepts the same setting. +The delay is a nonnegative integer in seconds. Its default is zero, preserving +existing behavior and stored options. It applies to contended workflow, +activity, signal and child-completion queue jobs. It does not change lock +expiry, semaphore ownership, or activity exception backoff. + +The setting is saved with the run. Children and continue-as-new inherit it +when no explicit `WorkflowOptions` is supplied. An explicit options object +replaces inherited options, including resetting this delay to zero if omitted. + +Laravel counts queue reservations as attempts even when middleware releases +the job before its handler executes. A positive delay reduces rapid retries; +it does not make a low `$tries`, `--tries`, or database queue attempt limit +safe for arbitrary contention. Choose the delay and retry budget for the +expected lock duration. Keep queue visibility and worker timeout settings +consistent with activity duration. + +This option belongs to the legacy API. `Workflow\V2\Workflow` uses durable +task claims and does not use this overlap middleware. diff --git a/src/Activity.php b/src/Activity.php index 4d5b163f6..64f137840 100644 --- a/src/Activity.php +++ b/src/Activity.php @@ -131,7 +131,8 @@ public function middleware() new WithoutOverlappingMiddleware( $this->storedWorkflow->id, WithoutOverlappingMiddleware::ACTIVITY, - 0, + $this->storedWorkflow->workflowOptions() +->overlapReleaseAfter, $this->timeout ), new ActivityMiddleware(), diff --git a/src/ChildWorkflow.php b/src/ChildWorkflow.php index 58d6e6ad0..c13d7ac4a 100644 --- a/src/ChildWorkflow.php +++ b/src/ChildWorkflow.php @@ -75,7 +75,13 @@ public function handle() public function middleware() { return [ - new WithoutOverlappingMiddleware($this->parentWorkflow->id, WithoutOverlappingMiddleware::ACTIVITY, 0, 15), + new WithoutOverlappingMiddleware( + $this->parentWorkflow->id, + WithoutOverlappingMiddleware::ACTIVITY, + $this->parentWorkflow->workflowOptions() +->overlapReleaseAfter, + 15 + ), ]; } diff --git a/src/ChildWorkflowStub.php b/src/ChildWorkflowStub.php index b2f2d8803..924e5d62b 100644 --- a/src/ChildWorkflowStub.php +++ b/src/ChildWorkflowStub.php @@ -121,7 +121,7 @@ public static function make($workflow, ...$arguments): PromiseInterface if (! $hasOptions) { $options = $context->storedWorkflow->workflowOptions(); - if ($options->connection !== null || $options->queue !== null) { + if ($options->connection !== null || $options->queue !== null || $options->overlapReleaseAfter !== 0) { $arguments[] = $options; } } diff --git a/src/Signal.php b/src/Signal.php index 4c2d679d6..4935f7153 100644 --- a/src/Signal.php +++ b/src/Signal.php @@ -46,7 +46,8 @@ public function middleware() new WithoutOverlappingMiddleware( $this->storedWorkflow->id, WithoutOverlappingMiddleware::WORKFLOW, - 0, + $this->storedWorkflow->workflowOptions() +->overlapReleaseAfter, 15 ), ]; diff --git a/src/Traits/Continues.php b/src/Traits/Continues.php index c2684733e..f8cc5c2e4 100644 --- a/src/Traits/Continues.php +++ b/src/Traits/Continues.php @@ -67,7 +67,7 @@ public static function continueAsNew(...$arguments): PromiseInterface if (! collect($arguments)->contains(static fn ($argument): bool => $argument instanceof WorkflowOptions)) { $options = $context->storedWorkflow->workflowOptions(); - if ($options->connection !== null || $options->queue !== null) { + if ($options->connection !== null || $options->queue !== null || $options->overlapReleaseAfter !== 0) { $arguments[] = $options; } } diff --git a/src/Workflow.php b/src/Workflow.php index c410328ff..aea55cce7 100644 --- a/src/Workflow.php +++ b/src/Workflow.php @@ -167,6 +167,8 @@ public function children(): array public function middleware() { + $releaseAfter = $this->storedWorkflow->workflowOptions() +->overlapReleaseAfter; $parentWorkflow = $this->storedWorkflow->parents() ->wherePivot('parent_index', '!=', StoredWorkflow::CONTINUE_PARENT_INDEX) ->wherePivot('parent_index', '!=', StoredWorkflow::ACTIVE_WORKFLOW_INDEX) @@ -174,11 +176,21 @@ public function middleware() if ($parentWorkflow) { return [ - new WithoutOverlappingMiddleware($parentWorkflow->id, WithoutOverlappingMiddleware::ACTIVITY, 0, 15), + new WithoutOverlappingMiddleware( + $parentWorkflow->id, + WithoutOverlappingMiddleware::ACTIVITY, + $releaseAfter, + 15 + ), ]; } return [ - new WithoutOverlappingMiddleware($this->storedWorkflow->id, WithoutOverlappingMiddleware::WORKFLOW, 0, 15), + new WithoutOverlappingMiddleware( + $this->storedWorkflow->id, + WithoutOverlappingMiddleware::WORKFLOW, + $releaseAfter, + 15 + ), ]; } diff --git a/src/WorkflowMetadata.php b/src/WorkflowMetadata.php index 96818fc89..efd036e50 100644 --- a/src/WorkflowMetadata.php +++ b/src/WorkflowMetadata.php @@ -18,7 +18,7 @@ public function __construct( /** * @return array{ * arguments: array, - * options: array{connection: ?string, queue: ?string} + * options: array{connection: ?string, queue: ?string, overlapReleaseAfter?: int} * } */ public function toArray(): array @@ -28,6 +28,9 @@ public function toArray(): array 'options' => [ 'connection' => $this->options->connection, 'queue' => $this->options->queue, + ...($this->options->overlapReleaseAfter === 0 ? [] : [ + 'overlapReleaseAfter' => $this->options->overlapReleaseAfter, + ]), ], ]; } diff --git a/src/WorkflowOptions.php b/src/WorkflowOptions.php index cccc96ede..4685cf77c 100644 --- a/src/WorkflowOptions.php +++ b/src/WorkflowOptions.php @@ -4,19 +4,38 @@ namespace Workflow; +use InvalidArgumentException; + final class WorkflowOptions { public function __construct( public readonly ?string $connection = null, public readonly ?string $queue = null, + public readonly int $overlapReleaseAfter = 0, ) { + if ($overlapReleaseAfter < 0) { + throw new InvalidArgumentException('overlapReleaseAfter must be a nonnegative number of seconds.'); + } + } + + public function __unserialize(array $data): void + { + // Options serialized before this setting was introduced keep their zero delay. + $options = self::set($data); + $this->connection = $options->connection; + $this->queue = $options->queue; + $this->overlapReleaseAfter = $options->overlapReleaseAfter; } /** - * @param array{connection?: string|null, queue?: string|null} $options + * @param array{connection?: string|null, queue?: string|null, overlapReleaseAfter?: int} $options */ public static function set(array $options): self { - return new self($options['connection'] ?? null, $options['queue'] ?? null); + return new self( + $options['connection'] ?? null, + $options['queue'] ?? null, + array_key_exists('overlapReleaseAfter', $options) ? $options['overlapReleaseAfter'] : 0, + ); } } diff --git a/tests/Fixtures/TestOverlapDelayWorkflow.php b/tests/Fixtures/TestOverlapDelayWorkflow.php new file mode 100644 index 000000000..a35b1dc94 --- /dev/null +++ b/tests/Fixtures/TestOverlapDelayWorkflow.php @@ -0,0 +1,17 @@ +assertSame(1, WorkflowStub::getContext()->index); } - public function testUsesParentContextForInheritedWorkflowOptions(): void + public static function inheritedWorkflowOptions(): array { + return [ + 'routing' => ['sync', 'parent-queue', 0], + 'delay only' => [null, null, 7], + ]; + } + + #[DataProvider('inheritedWorkflowOptions')] + public function testUsesParentContextForInheritedWorkflowOptions( + ?string $connection, + ?string $queue, + int $delay + ): void { $childContextStoredWorkflow = Mockery::mock(); $childContextStoredWorkflow->shouldNotReceive('workflowOptions'); @@ -359,7 +372,7 @@ public function testUsesParentContextForInheritedWorkflowOptions(): void ->andReturn(null); $parentStoredWorkflow->shouldReceive('workflowOptions') ->once() - ->andReturn(new WorkflowOptions('sync', 'parent-queue')); + ->andReturn(new WorkflowOptions($connection, $queue, $delay)); $childWorkflow = Mockery::mock(); $childWorkflow->shouldReceive('running') @@ -371,7 +384,7 @@ public function testUsesParentContextForInheritedWorkflowOptions(): void $childWorkflow->shouldReceive('startAsChild') ->once() ->withArgs( - static function (...$arguments) use ($parentStoredWorkflow): bool { + static function (...$arguments) use ($parentStoredWorkflow, $connection, $queue, $delay): bool { if (count($arguments) !== 4) { return false; } @@ -381,8 +394,9 @@ static function (...$arguments) use ($parentStoredWorkflow): bool { return $parentWorkflow === $parentStoredWorkflow && $index === 0 && $options instanceof WorkflowOptions - && $options->connection === 'sync' - && $options->queue === 'parent-queue'; + && $options->connection === $connection + && $options->queue === $queue + && $options->overlapReleaseAfter === $delay; } ); diff --git a/tests/Unit/WorkflowOptionsTest.php b/tests/Unit/WorkflowOptionsTest.php new file mode 100644 index 000000000..4c705f4dd --- /dev/null +++ b/tests/Unit/WorkflowOptionsTest.php @@ -0,0 +1,81 @@ +toArray()); + $this->assertSame(['argument'], $restored->arguments); + $this->assertSame(5, $restored->options->overlapReleaseAfter); + $this->assertEquals($options, unserialize(serialize($options))); + } + + public function testExistingSerializedOptionsKeepTheZeroDefault(): void + { + $options = unserialize('O:24:"Workflow\\WorkflowOptions":2:{s:10:"connection";N;s:5:"queue";N;}'); + $this->assertInstanceOf(WorkflowOptions::class, $options); + $this->assertSame(0, $options->overlapReleaseAfter); + $this->assertSame(0, WorkflowOptions::set([])->overlapReleaseAfter); + $metadata = WorkflowMetadata::fromSerializedArguments([ + 'arguments' => [], + 'options' => [ + 'connection' => 'database', + 'queue' => 'default', + ], + ]); + $this->assertSame(0, $metadata->options->overlapReleaseAfter); + $this->assertSame([ + 'connection' => 'database', + 'queue' => 'default', + ], $metadata->toArray()['options'],); + } + + public function testConstructorRejectsNegativeDelay(): void + { + $this->expectException(InvalidArgumentException::class); + new WorkflowOptions(overlapReleaseAfter: -1); + } + + public function testExplicitOptionsReplaceTheInheritedDelay(): void + { + $inherited = new WorkflowOptions(overlapReleaseAfter: 7); + $metadata = WorkflowMetadata::fromStartArguments(['argument'], $inherited); + $this->assertSame(7, $metadata->options->overlapReleaseAfter); + $metadata = WorkflowMetadata::fromStartArguments(['argument', new WorkflowOptions()], $inherited); + $this->assertSame(0, $metadata->options->overlapReleaseAfter); + $this->assertSame(['argument'], $metadata->arguments); + } + + public static function invalidDelays(): array + { + return [ + 'negative' => [-1, InvalidArgumentException::class], + 'fraction' => [0.5, TypeError::class], + 'string' => ['5', TypeError::class], + 'boolean' => [true, TypeError::class], + 'null' => [null, TypeError::class], + ]; + } + + #[DataProvider('invalidDelays')] + public function testRejectsInvalidPersistedDelay(mixed $delay, string $exception): void + { + $this->expectException($exception); + WorkflowOptions::set([ + 'overlapReleaseAfter' => $delay, + ]); + } +} diff --git a/tests/Unit/WorkflowOverlapReleaseTest.php b/tests/Unit/WorkflowOverlapReleaseTest.php new file mode 100644 index 000000000..603adbad3 --- /dev/null +++ b/tests/Unit/WorkflowOverlapReleaseTest.php @@ -0,0 +1,202 @@ +bigIncrements('id'); + $table->string('queue') + ->index(); + $table->longText('payload'); + $table->unsignedTinyInteger('attempts'); + $table->unsignedInteger('reserved_at') + ->nullable(); + $table->unsignedInteger('available_at'); + $table->unsignedInteger('created_at'); + }); + } + + public static function releaseDelays(): array + { + return [ + 'zero' => [0], + 'positive' => [5], + ]; + } + + public static function otherContendedJobs(): array + { + return [ + 'child workflow' => ['child', 5], + 'child result' => ['result', 7], + 'activity' => ['activity', 5], + 'signal' => ['signal', 5], + ]; + } + + #[DataProvider('otherContendedJobs')] + public function testRelatedJobsUseTheSelectedDelayWithoutChangingTheLockOwner(string $kind, int $delay): void + { + Carbon::setTestNow(now()->startOfSecond()); + config([ + 'queue.default' => 'database', + ]); + $stored = $this->storedWorkflowWithDelay(5); + $parent = $this->storedWorkflowWithDelay(7); + $stored->parents() + ->attach($parent, [ + 'parent_index' => 0, + 'parent_now' => now(), + ]); + $job = match ($kind) { + 'child' => new TestOverlapDelayWorkflow($stored), + 'result' => new ChildWorkflow(0, now()->toDateTimeString(), $stored, 'result', $parent), + 'activity' => new TestOtherActivity(0, now()->toDateTimeString(), $stored), + 'signal' => new Signal($stored), + }; + $middleware = $job->middleware()[0]; + $owner = in_array($kind, ['child', 'result'], true) ? $parent->id : $stored->id; + $this->assertSame((string) $owner, $middleware->key); + $this->assertSame($delay, $middleware->releaseAfter); + $this->assertSame( + $kind === 'signal' ? WithoutOverlappingMiddleware::WORKFLOW : WithoutOverlappingMiddleware::ACTIVITY, + $middleware->type + ); + $this->assertSame($kind === 'activity' ? $job->timeout : 15, $middleware->expiresAfter); + + $lock = new WithoutOverlappingMiddleware($owner, WithoutOverlappingMiddleware::WORKFLOW); + $holder = new \stdClass(); + $this->assertTrue($lock->lock($holder)); + $queue = Queue::connection('database'); + $queue->push($job); + /** @var Worker $worker */ + $worker = $this->app->make('queue.worker'); + $worker->setCache($this->app->make('cache.store')); + + try { + $blocked = $queue->pop(); + $this->assertNotNull($blocked); + $payload = $blocked->getRawBody(); + $worker->process('database', $blocked, new WorkerOptions()); + $this->assertTrue($blocked->isReleased()); + $this->assertFalse($blocked->hasFailed()); + $this->assertTrue($queue->pop() === null); + $this->assertTrue($lock->unlock($holder)); + Carbon::setTestNow(now()->addSeconds($delay)); + $ready = $queue->pop(); + $this->assertNotNull($ready); + $this->assertSame(2, $ready->attempts()); + $this->assertSame($payload, $ready->getRawBody()); + $ready->delete(); + } finally { + Carbon::setTestNow(); + } + } + + #[DataProvider('releaseDelays')] + public function testDatabaseQueueContentionPreservesTheWorkflowAndDelaysReadmission(int $delay): void + { + Carbon::setTestNow(now()->startOfSecond()); + config([ + 'queue.default' => 'database', + ]); + $stub = WorkflowStub::make(TestOverlapDelayWorkflow::class); + $stored = StoredWorkflow::findOrFail($stub->id()); + $stored->update([ + 'arguments' => Serializer::serialize([ + 'arguments' => [], + 'options' => [ + 'connection' => 'database', + 'overlapReleaseAfter' => $delay, + ], + ]), + 'status' => WorkflowPendingStatus::class, + ]); + $workflow = new TestOverlapDelayWorkflow($stored); + $queue = Queue::connection('database'); + $queue->push($workflow); + /** @var Worker $worker */ + $worker = $this->app->make('queue.worker'); + $worker->setCache($this->app->make('cache.store')); + + // Two outstanding activity locks reproduce a parallel activity group. + $lock = new WithoutOverlappingMiddleware($stub->id(), WithoutOverlappingMiddleware::ACTIVITY); + $firstActivity = new \stdClass(); + $secondActivity = new \stdClass(); + $this->assertTrue($lock->lock($firstActivity)); + $this->assertTrue($lock->lock($secondActivity)); + + try { + $blocked = $queue->pop(); + $this->assertNotNull($blocked); + $payload = $blocked->getRawBody(); + $this->assertSame(1, $blocked->attempts()); + $worker->process('database', $blocked, new WorkerOptions()); + $this->assertTrue($blocked->isReleased()); + $this->assertFalse($blocked->hasFailed()); + $this->assertSame(0, $stub->exceptions()->count()); + $this->assertFalse($lock->unlock($firstActivity)); + $this->assertTrue($lock->unlock($secondActivity)); + + if ($delay > 0) { + $this->assertTrue($queue->pop() === null, 'A released job must not be immediately available.'); + Carbon::setTestNow(now()->addSeconds($delay - 1)); + $this->assertNull($queue->pop()); + Carbon::setTestNow(now()->addSecond()); + } + + $ready = $queue->pop(); + $this->assertNotNull($ready); + $this->assertSame(2, $ready->attempts()); + $this->assertSame($payload, $ready->getRawBody()); + $worker->process('database', $ready, new WorkerOptions()); + $this->assertFalse($ready->isReleased()); + $this->assertFalse($ready->hasFailed()); + $this->assertSame(0, $queue->size()); + $this->assertTrue($stub->completed()); + $this->assertSame('completed', $stub->output()); + } finally { + Carbon::setTestNow(); + } + } + + private function storedWorkflowWithDelay(int $delay): StoredWorkflow + { + return StoredWorkflow::create([ + 'class' => TestOverlapDelayWorkflow::class, + 'arguments' => Serializer::serialize([ + 'arguments' => [], + 'options' => [ + 'connection' => 'database', + 'overlapReleaseAfter' => $delay, + ], + ]), + 'status' => WorkflowPendingStatus::class, + ]); + } +} diff --git a/tests/Unit/WorkflowTest.php b/tests/Unit/WorkflowTest.php index 580179d11..e0218aeac 100644 --- a/tests/Unit/WorkflowTest.php +++ b/tests/Unit/WorkflowTest.php @@ -9,6 +9,7 @@ use Illuminate\Support\Carbon; use Illuminate\Support\Facades\Event; use Mockery; +use PHPUnit\Framework\Attributes\DataProvider; use ReflectionMethod; use Tests\Fixtures\TestActivity; use Tests\Fixtures\TestChildWorkflow; @@ -525,15 +526,25 @@ public function testContinueAsNewWithParentWorkflow(): void $this->assertSame(1, $storedWorkflow->continuedWorkflows()->count()); } - public function testContinueAsNewCarriesWorkflowOptions(): void + public static function inheritedWorkflowOptions(): array + { + return [ + 'routing' => ['sync', 'default', 0], + 'delay only' => [null, null, 7], + ]; + } + + #[DataProvider('inheritedWorkflowOptions')] + public function testContinueAsNewCarriesWorkflowOptions(?string $connection, ?string $queue, int $delay): void { $storedWorkflow = StoredWorkflow::create([ 'class' => TestContinueAsNewWorkflow::class, 'arguments' => Serializer::serialize([ 'arguments' => [0, 3], 'options' => [ - 'connection' => 'sync', - 'queue' => 'default', + 'connection' => $connection, + 'queue' => $queue, + 'overlapReleaseAfter' => $delay, ], ]), 'status' => WorkflowPendingStatus::class, @@ -554,8 +565,9 @@ public function testContinueAsNewCarriesWorkflowOptions(): void ->first(); $this->assertNotNull($continuedWorkflow); - $this->assertSame('sync', $continuedWorkflow->workflowOptions()->connection); - $this->assertSame('default', $continuedWorkflow->workflowOptions()->queue); + $this->assertSame($connection, $continuedWorkflow->workflowOptions()->connection); + $this->assertSame($queue, $continuedWorkflow->workflowOptions()->queue); + $this->assertSame($delay, $continuedWorkflow->workflowOptions()->overlapReleaseAfter); } public function testRedeliveredJobResumesFromRunningState(): void