From 24a18108b38f0b7fa320afed4ee053ebf2a65940 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:13:54 +0000 Subject: [PATCH 1/2] test: cover inbox receives after grouped activities --- tests/Feature/V2/V2InboxAfterParallelTest.php | 114 ++++++++++++++++++ .../V2/TestInboxAfterParallelWorkflow.php | 44 +++++++ .../V2/inbox_after_parallel_cold_read.php | 23 ++++ 3 files changed, 181 insertions(+) create mode 100644 tests/Feature/V2/V2InboxAfterParallelTest.php create mode 100644 tests/Fixtures/V2/TestInboxAfterParallelWorkflow.php create mode 100644 tests/Fixtures/V2/inbox_after_parallel_cold_read.php diff --git a/tests/Feature/V2/V2InboxAfterParallelTest.php b/tests/Feature/V2/V2InboxAfterParallelTest.php new file mode 100644 index 000000000..f6fb2adbb --- /dev/null +++ b/tests/Feature/V2/V2InboxAfterParallelTest.php @@ -0,0 +1,114 @@ +set('workflows.v2.task_dispatch_mode', 'poll'); + $stub = WorkflowStub::make(TestInboxAfterParallelWorkflow::class, ($parallel ? 'after-all-' : 'after-sequential-') . $admission); + $stub->start($parallel); + $this->drain(); + $run = WorkflowRun::query()->findOrFail($stub->runId()); + $send = static fn () => (new MessageService())->sendMessage($run, MessageChannel::WorkflowMessage, $run->workflow_instance_id, 'j1', 'chat'); + if ($admission === 0) { + $send(); + } + $stub->signal('go'); + if ($admission === 2) { + for ($turn = 0; $turn < 20; $turn++) { + if (ActivityExecution::query()->where('workflow_run_id', $run->id)->where('status', 'completed')->count() === 2) { + break; + } + self::assertTrue($this->runNextTask(), 'Two real activities must complete before admission.'); + } + } + if ($admission !== 0) { + $send(); + } + $this->drain(); + self::assertSame(2, ActivityExecution::query()->where('workflow_run_id', $run->id)->where('status', 'completed')->count()); + $first = $stub->refresh()->query('received'); + $before = $this->rows(); + $reader = new Process([PHP_BINARY, __DIR__ . '/../../Fixtures/V2/inbox_after_parallel_cold_read.php'], dirname(__DIR__, 3), [ + 'INBOX_PARALLEL_RUN_ID' => $run->id, + ]); + $reader->setTimeout(30); + $reader->mustRun(); + $cold = json_decode($reader->getOutput(), true, 512, JSON_THROW_ON_ERROR)['received']; + self::assertSame($before, $this->rows(), 'A fresh-process query cannot change any durable row.'); + $stub->signal('done'); + $this->drain(); + self::assertTrue($stub->refresh()->completed()); + self::assertSame(['Hello, first!', 'Hello, second!'], $stub->output()['activities']); + self::assertSame(['j1', 'j1', 'j1'], [$first, $cold, $stub->output()['received']]); + } + + public static function executionModes(): iterable + { + foreach ([0 => 'before signal', 1 => 'before activities', 2 => 'after activities'] as $admission => $name) { + yield 'sequential ' . $name => [false, $admission]; + yield 'parallel ' . $name => [true, $admission]; + } + } + + private function drain(): void + { + for ($turn = 0; $turn < 50; $turn++) { + if (! $this->runNextTask()) { + return; + } + } + self::fail('The bounded source fixture did not reach a wait.'); + } + + private function runNextTask(): bool + { + $task = WorkflowTask::query()->where('status', TaskStatus::Ready->value)->orderBy('created_at')->first(); + if ($task === null || ($task->available_at !== null && $task->available_at->isFuture())) { + return false; + } + $job = match ($task->task_type) { + TaskType::Workflow => new RunWorkflowTask($task->id), + TaskType::Activity => new RunActivityTask($task->id), + TaskType::Timer => new RunTimerTask($task->id), + }; + $this->app->call([$job, 'handle']); + return true; + } + + private function rows(): array + { + $rows = []; + foreach (DB::getSchemaBuilder()->getTableListing() as $table) { + $values = DB::table($table)->get()->map(static fn ($row): string => json_encode($row, JSON_THROW_ON_ERROR))->all(); + sort($values); + $rows[$table] = $values; + } + ksort($rows); + return $rows; + } +} diff --git a/tests/Fixtures/V2/TestInboxAfterParallelWorkflow.php b/tests/Fixtures/V2/TestInboxAfterParallelWorkflow.php new file mode 100644 index 000000000..84abb5554 --- /dev/null +++ b/tests/Fixtures/V2/TestInboxAfterParallelWorkflow.php @@ -0,0 +1,44 @@ + activity(TestGreetingActivity::class, 'first'), + static fn () => activity(TestGreetingActivity::class, 'second'), + ]); + } else { + $results = [activity(TestGreetingActivity::class, 'first'), activity(TestGreetingActivity::class, 'second')]; + } + $this->received = $this->inbox('chat')->receiveOne()?->payload_reference; + await('done'); + + return ['activities' => $results, 'received' => $this->received]; + } + + #[QueryMethod] + public function received(): ?string + { + return $this->received; + } +} diff --git a/tests/Fixtures/V2/inbox_after_parallel_cold_read.php b/tests/Fixtures/V2/inbox_after_parallel_cold_read.php new file mode 100644 index 000000000..577103180 --- /dev/null +++ b/tests/Fixtures/V2/inbox_after_parallel_cold_read.php @@ -0,0 +1,23 @@ + false, + 'extra' => [ + 'providers' => [\Workflow\Providers\WorkflowServiceProvider::class], + ], +]); +$app->make(\Illuminate\Contracts\Console\Kernel::class)->bootstrap(); +try { + $result = \Workflow\V2\WorkflowStub::loadRun((string) getenv('INBOX_PARALLEL_RUN_ID'))->query('received'); + echo json_encode(['received' => $result], JSON_THROW_ON_ERROR); +} finally { + $app->make('db')->purge(); +} From 708d3748edee5677d646de294fff96ab07a663f4 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:23:55 +0000 Subject: [PATCH 2/2] style: normalize grouped inbox regression fixtures --- tests/Feature/V2/V2InboxAfterParallelTest.php | 44 +++++++++++++++---- .../V2/TestInboxAfterParallelWorkflow.php | 13 ++++-- .../V2/inbox_after_parallel_cold_read.php | 7 ++- 3 files changed, 50 insertions(+), 14 deletions(-) diff --git a/tests/Feature/V2/V2InboxAfterParallelTest.php b/tests/Feature/V2/V2InboxAfterParallelTest.php index f6fb2adbb..8c4b5db7d 100644 --- a/tests/Feature/V2/V2InboxAfterParallelTest.php +++ b/tests/Feature/V2/V2InboxAfterParallelTest.php @@ -28,19 +28,32 @@ final class V2InboxAfterParallelTest extends TestCase public function testPendingReceiveAfterActivitiesSurvivesColdReplay(bool $parallel, int $admission): void { Queue::fake(); - config()->set('workflows.v2.task_dispatch_mode', 'poll'); - $stub = WorkflowStub::make(TestInboxAfterParallelWorkflow::class, ($parallel ? 'after-all-' : 'after-sequential-') . $admission); + config() + ->set('workflows.v2.task_dispatch_mode', 'poll'); + $stub = WorkflowStub::make( + TestInboxAfterParallelWorkflow::class, + ($parallel ? 'after-all-' : 'after-sequential-') . $admission + ); $stub->start($parallel); $this->drain(); $run = WorkflowRun::query()->findOrFail($stub->runId()); - $send = static fn () => (new MessageService())->sendMessage($run, MessageChannel::WorkflowMessage, $run->workflow_instance_id, 'j1', 'chat'); + $send = static fn () => (new MessageService())->sendMessage( + $run, + MessageChannel::WorkflowMessage, + $run->workflow_instance_id, + 'j1', + 'chat' + ); if ($admission === 0) { $send(); } $stub->signal('go'); if ($admission === 2) { for ($turn = 0; $turn < 20; $turn++) { - if (ActivityExecution::query()->where('workflow_run_id', $run->id)->where('status', 'completed')->count() === 2) { + if (ActivityExecution::query()->where('workflow_run_id', $run->id)->where( + 'status', + 'completed' + )->count() === 2) { break; } self::assertTrue($this->runNextTask(), 'Two real activities must complete before admission.'); @@ -50,10 +63,17 @@ public function testPendingReceiveAfterActivitiesSurvivesColdReplay(bool $parall $send(); } $this->drain(); - self::assertSame(2, ActivityExecution::query()->where('workflow_run_id', $run->id)->where('status', 'completed')->count()); - $first = $stub->refresh()->query('received'); + self::assertSame( + 2, + ActivityExecution::query()->where('workflow_run_id', $run->id)->where('status', 'completed')->count() + ); + $first = $stub->refresh() + ->query('received'); $before = $this->rows(); - $reader = new Process([PHP_BINARY, __DIR__ . '/../../Fixtures/V2/inbox_after_parallel_cold_read.php'], dirname(__DIR__, 3), [ + $reader = new Process([PHP_BINARY, __DIR__ . '/../../Fixtures/V2/inbox_after_parallel_cold_read.php'], dirname( + __DIR__, + 3 + ), [ 'INBOX_PARALLEL_RUN_ID' => $run->id, ]); $reader->setTimeout(30); @@ -69,7 +89,11 @@ public function testPendingReceiveAfterActivitiesSurvivesColdReplay(bool $parall public static function executionModes(): iterable { - foreach ([0 => 'before signal', 1 => 'before activities', 2 => 'after activities'] as $admission => $name) { + foreach ([ + 0 => 'before signal', + 1 => 'before activities', + 2 => 'after activities', + ] as $admission => $name) { yield 'sequential ' . $name => [false, $admission]; yield 'parallel ' . $name => [true, $admission]; } @@ -104,7 +128,9 @@ private function rows(): array { $rows = []; foreach (DB::getSchemaBuilder()->getTableListing() as $table) { - $values = DB::table($table)->get()->map(static fn ($row): string => json_encode($row, JSON_THROW_ON_ERROR))->all(); + $values = DB::table($table)->get()->map( + static fn ($row): string => json_encode($row, JSON_THROW_ON_ERROR) + )->all(); sort($values); $rows[$table] = $values; } diff --git a/tests/Fixtures/V2/TestInboxAfterParallelWorkflow.php b/tests/Fixtures/V2/TestInboxAfterParallelWorkflow.php index 84abb5554..a4a7e8cf5 100644 --- a/tests/Fixtures/V2/TestInboxAfterParallelWorkflow.php +++ b/tests/Fixtures/V2/TestInboxAfterParallelWorkflow.php @@ -28,12 +28,19 @@ public function handle(bool $parallel): array static fn () => activity(TestGreetingActivity::class, 'second'), ]); } else { - $results = [activity(TestGreetingActivity::class, 'first'), activity(TestGreetingActivity::class, 'second')]; + $results = [ + activity(TestGreetingActivity::class, 'first'), + activity(TestGreetingActivity::class, 'second'), + ]; } - $this->received = $this->inbox('chat')->receiveOne()?->payload_reference; + $this->received = $this->inbox('chat') + ->receiveOne()?->payload_reference; await('done'); - return ['activities' => $results, 'received' => $this->received]; + return [ + 'activities' => $results, + 'received' => $this->received, + ]; } #[QueryMethod] diff --git a/tests/Fixtures/V2/inbox_after_parallel_cold_read.php b/tests/Fixtures/V2/inbox_after_parallel_cold_read.php index 577103180..0489dfb7a 100644 --- a/tests/Fixtures/V2/inbox_after_parallel_cold_read.php +++ b/tests/Fixtures/V2/inbox_after_parallel_cold_read.php @@ -17,7 +17,10 @@ $app->make(\Illuminate\Contracts\Console\Kernel::class)->bootstrap(); try { $result = \Workflow\V2\WorkflowStub::loadRun((string) getenv('INBOX_PARALLEL_RUN_ID'))->query('received'); - echo json_encode(['received' => $result], JSON_THROW_ON_ERROR); + echo json_encode([ + 'received' => $result, + ], JSON_THROW_ON_ERROR); } finally { - $app->make('db')->purge(); + $app->make('db') + ->purge(); }