diff --git a/.github/workflows/php.yml b/.github/workflows/php.yml index 34b8d8465..f2af6f729 100644 --- a/.github/workflows/php.yml +++ b/.github/workflows/php.yml @@ -394,6 +394,20 @@ jobs: tests/Unit/V2/ProjectionSqliteContentionTest.php \ --fail-on-warning --log-junit build/test-results/pr-database-queue-recovery.xml + - name: Run legacy overlap releases on a real database queue + env: + DB_CONNECTION: sqlite + DB_DATABASE: ${{ runner.temp }}/workflow-overlap-release.sqlite + CACHE_DRIVER: array + CACHE_STORE: array + QUEUE_CONNECTION: sync + REDIS_HOST: '' + run: | + touch "$DB_DATABASE" + vendor/bin/phpunit tests/Unit/WorkflowOverlapReleaseTest.php \ + tests/Unit/WorkflowOptionsTest.php \ + tests/Unit/V2/LegacyOverlapOptionsBoundaryTest.php --testdox --fail-on-warning + - name: Run retained execution authority and process-loss controls run: >- vendor/bin/phpunit tests/Unit/V2/EmbeddedContinuationTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 55044899b..eef00a9ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## 2.11.0 - 2026-10-22 + +- Add the opt-in `WorkflowOptions::overlapReleaseAfter` delay for legacy + workflow, activity, signal and child-completion jobs contending on overlap + locks. Preserve the zero default, stored options, lock ownership and expiry. +- Preserve the delay through child inheritance and continue-as-new. Queue + releases still count toward Laravel attempt limits. + ## 2.10.2 - 2026-10-22 - Preserve recorded activity lease expiry and worker deregistration in canonical diff --git a/composer.json b/composer.json index 72c650394..a8f9666df 100644 --- a/composer.json +++ b/composer.json @@ -87,7 +87,7 @@ "dev-main": "2.0.x-dev" }, "durable-workflow": { - "product-train": "2.10.2", + "product-train": "2.11.0", "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json", "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json" }, diff --git a/docs/architecture/routing-precedence.md b/docs/architecture/routing-precedence.md index ed6728b32..22a4893ef 100644 --- a/docs/architecture/routing-precedence.md +++ b/docs/architecture/routing-precedence.md @@ -133,10 +133,11 @@ The contract authorities are: `RoutingResolver::activityConnection`, and `RoutingResolver::activityQueue`. No other class is allowed to compute `(connection, queue)` from the same inputs. -- **`WorkflowOptions`** — the durable wire value for workflow start - routing overrides. The type carries only `?string $connection` and - `?string $queue`. Any additional fields on `WorkflowOptions` are a - protocol change. +- **`WorkflowOptions`** — workflow start routing overrides are + `?string $connection` and `?string $queue`. The optional nonnegative + `int $overlapReleaseAfter` controls only the legacy `Workflow\\Workflow` + queue middleware. It defaults to zero, is omitted from stored metadata at + that default, and does not change V2 routing or the portable service protocol. - **`ActivityOptions`** — the durable wire value for activity scheduling overrides. Routing fields on `ActivityOptions` are `?string $connection` and `?string $queue`; retry and timeout 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 44e116b9b..f22e488e7 100644 --- a/src/Activity.php +++ b/src/Activity.php @@ -138,7 +138,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 631b3c5a3..3485fcb54 100644 --- a/src/ChildWorkflow.php +++ b/src/ChildWorkflow.php @@ -73,7 +73,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 b1e40a907..1a99a54c2 100644 --- a/src/Workflow.php +++ b/src/Workflow.php @@ -174,6 +174,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) @@ -181,11 +183,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/V2/LegacyOverlapOptionsBoundaryTest.php b/tests/Unit/V2/LegacyOverlapOptionsBoundaryTest.php new file mode 100644 index 000000000..418bcc644 --- /dev/null +++ b/tests/Unit/V2/LegacyOverlapOptionsBoundaryTest.php @@ -0,0 +1,113 @@ + ['start'], + 'signal with start' => ['signal'], + 'child start' => ['child'], + ]; + } + + #[DataProvider('entryPoints')] + public function testLegacyDelayDoesNotChangeV2ArgumentsRoutingOrPortableShape(string $entry): void + { + Queue::fake(); + config([ + 'workflows.v2.task_dispatch_mode' => 'queue', + ]); + $seen = []; + $snapshots = []; + foreach ([0, 5] as $delay) { + $workflow = WorkflowStub::make( + $entry === 'child' ? LegacyOverlapBoundaryParent::class : LegacyOverlapBoundaryWorkflow::class, + ); + $options = new WorkflowOptions('redis', 'legacy-boundary', $delay); + if ($entry === 'signal') { + $this->assertTrue( + $workflow->signalWithStart('poke', ['signal-value'], 'argument', $options)->accepted() + ); + } elseif ($entry === 'child') { + $workflow->start('argument', $delay, $options); + $parentTask = WorkflowTask::query()->where('workflow_run_id', $workflow->runId())->sole(); + $this->assertTrue(app(DefaultWorkflowTaskBridge::class)->execute($parentTask->id)['executed']); + } else { + $workflow->start('argument', $options); + } + $run = WorkflowRun::query()->where('workflow_class', LegacyOverlapBoundaryWorkflow::class) + ->whereNotIn('id', $seen) + ->sole(); + $seen[] = $run->id; + $task = WorkflowTask::query()->where('workflow_run_id', $run->id)->sole(); + $this->assertSame(['argument'], Serializer::unserializeWithCodec($run->payload_codec, $run->arguments)); + $this->assertSame('redis', $run->connection); + $this->assertSame('legacy-boundary', $run->queue); + $this->assertSame($run->connection, $task->connection); + $this->assertSame($run->queue, $task->queue); + $this->assertTrue(app(DefaultWorkflowTaskBridge::class)->execute($task->id)['executed']); + $this->assertSame('argument', WorkflowStub::load($run->workflow_instance_id)->output()); + $task->refresh(); + $history = $run->historyEvents() + ->orderBy('sequence') + ->get(); + $portable = [$task->payload, ...$history->pluck('payload')->all()]; + $this->assertStringNotContainsString('overlapReleaseAfter', json_encode($portable, JSON_THROW_ON_ERROR)); + $snapshots[] = [ + 'task' => $this->shape($task->getAttributes()), + 'payload' => $this->shape($task->payload), + 'history' => $history->map(fn ($event): array => [ + $event->event_type->value, + $this->shape($event->payload), + ])->all(), + ]; + } + $this->assertSame($snapshots[0], $snapshots[1]); + } + + private function shape(mixed $value): mixed + { + return is_array($value) ? array_map($this->shape(...), $value) : get_debug_type($value); + } +} + +#[Type('legacy-overlap-boundary')] +#[Signal('poke')] +final class LegacyOverlapBoundaryWorkflow extends Workflow +{ + public function handle(string $argument): string + { + return $argument; + } +} + +#[Type('legacy-overlap-boundary-parent')] +final class LegacyOverlapBoundaryParent extends Workflow +{ + public function handle(string $argument, int $delay): mixed + { + return self::child( + LegacyOverlapBoundaryWorkflow::class, + $argument, + new WorkflowOptions('redis', 'legacy-boundary', $delay), + ); + } +} diff --git a/tests/Unit/V2/RoutingOptionsContractTest.php b/tests/Unit/V2/RoutingOptionsContractTest.php index 2dc192255..ca4a1f5fe 100644 --- a/tests/Unit/V2/RoutingOptionsContractTest.php +++ b/tests/Unit/V2/RoutingOptionsContractTest.php @@ -11,7 +11,7 @@ final class RoutingOptionsContractTest extends TestCase { - public function testWorkflowOptionsWireShapeIsLimitedToConnectionAndQueue(): void + public function testWorkflowOptionsSeparatesRoutingFromTheLegacyOverlapDelay(): void { $properties = array_map( static fn (ReflectionProperty $property): string => $property->getName(), @@ -20,11 +20,12 @@ public function testWorkflowOptionsWireShapeIsLimitedToConnectionAndQueue(): voi sort($properties); - $this->assertSame(['connection', 'queue'], $properties); + $this->assertSame(['connection', 'overlapReleaseAfter', 'queue'], $properties); $this->assertSame( [ 'connection' => 'redis', 'queue' => 'billing', + 'overlapReleaseAfter' => 0, ], get_object_vars(new WorkflowOptions(connection: 'redis', queue: 'billing')), ); 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..6ae53aa53 --- /dev/null +++ b/tests/Unit/WorkflowOverlapReleaseTest.php @@ -0,0 +1,183 @@ + [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 144e23ded..4862961b5 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; @@ -479,15 +480,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, @@ -508,8 +519,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