diff --git a/src/V2/Support/ActivityOutcomeRecorder.php b/src/V2/Support/ActivityOutcomeRecorder.php index 692cf0cc4..7ef6551cc 100644 --- a/src/V2/Support/ActivityOutcomeRecorder.php +++ b/src/V2/Support/ActivityOutcomeRecorder.php @@ -495,6 +495,7 @@ public static function record( $run, $parallelMetadataPath, $closedStatus, + lockHistoryForUpdate: true, ) ) { self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures'])); diff --git a/src/V2/Support/ActivityTimeoutEnforcer.php b/src/V2/Support/ActivityTimeoutEnforcer.php index ce8e0a605..0fb63ba3f 100644 --- a/src/V2/Support/ActivityTimeoutEnforcer.php +++ b/src/V2/Support/ActivityTimeoutEnforcer.php @@ -680,6 +680,7 @@ private static function recordTerminalTimeout( $run, $parallelMetadataPath, ActivityStatus::Failed, + lockHistoryForUpdate: true, ) ) { self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures'])); diff --git a/src/V2/Support/ParallelChildGroup.php b/src/V2/Support/ParallelChildGroup.php index 84be4ac9a..1e0ce4768 100644 --- a/src/V2/Support/ParallelChildGroup.php +++ b/src/V2/Support/ParallelChildGroup.php @@ -276,13 +276,15 @@ public static function shouldWakeParentOnChildClosure( public static function shouldWakeParentOnActivityClosure( WorkflowRun $parentRun, array $metadata, - ActivityStatus $closedActivityStatus + ActivityStatus $closedActivityStatus, + bool $lockHistoryForUpdate = false, ): bool { return self::shouldWakeParentOnClosure( $parentRun, self::normalizedPath($metadata), 'activity', $closedActivityStatus, + $lockHistoryForUpdate, ); } diff --git a/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php b/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php index 4cd3283e6..29e20aaf6 100644 --- a/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php +++ b/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php @@ -4,11 +4,14 @@ namespace Tests\Feature\V2; +use Illuminate\Database\Events\QueryExecuted; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; +use PHPUnit\Framework\Attributes\DataProvider; use RuntimeException; use Tests\TestCase; use Workflow\Serializers\Serializer; +use Workflow\V2\Contracts\ActivityTaskBridge; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; @@ -344,6 +347,165 @@ public function testConcurrentChildCompletionsSerializeOneBarrierWakeOnParentRun $this->assertSame(1, $this->openParentWorkflowTaskCount($parentRun)); } + public static function activityHistoryInterleavings(): array + { + $cases = []; + foreach ([false, true] as $named) { + foreach ([2, 6, 25] as $size) { + $cases[($named ? 'named' : 'default') . '-' . $size] = [$size, $named, false]; + } + $cases[($named ? 'named' : 'default') . '-nested'] = [6, $named, true]; + } + + return $cases; + } + + #[DataProvider('activityHistoryInterleavings')] + public function testActivityClosureReadsCurrentHistoryAfterAnEarlierSnapshot( + int $size, + bool $named, + bool $nested, + ): void { + if (DB::connection()->getDriverName() !== 'mysql') { + $this->markTestSkipped('This interleaving exercises MySQL repeatable-read snapshots.'); + } + + $originalStorage = config('workflows.storage.connection'); + $default = DB::getDefaultConnection(); + $reader = $named ? 'activity_barrier_reader' : $default; + $writer = 'activity_barrier_writer'; + config([ + 'database.connections.' . $writer => config('database.connections.' . $default), + ]); + if ($named) { + config([ + 'database.connections.' . $reader => config('database.connections.' . $default), + ]); + } + config([ + 'workflows.storage.connection' => $named ? $reader : null, + ]); + $observedSnapshot = false; + + try { + $readConnection = DB::connection($reader); + $writeConnection = DB::connection($writer); + $readConnection->statement('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ'); + $writeConnection->statement('SET SESSION innodb_lock_wait_timeout = 5'); + $this->assertNotSame( + $readConnection->scalar('SELECT CONNECTION_ID()'), + $writeConnection->scalar('SELECT CONNECTION_ID()'), + ); + + [$run, $attempts] = $this->stageActivityGroup($size, $nested); + $activities = $this->app->make(ActivityTaskBridge::class); + foreach (array_slice($attempts, 0, -2) as $attempt) { + $this->assertTrue($activities->complete($attempt, 'earlier sibling')['recorded']); + $this->assertSame(0, $this->openParentWorkflowTaskCount($run)); + } + $first = $attempts[$size - 2]; + $last = $attempts[$size - 1]; + $readStorage = config('workflows.storage.connection'); + + DB::listen(function (QueryExecuted $event) use ( + &$observedSnapshot, + $reader, + $writer, + $first, + $run, + $size, + $activities, + $readStorage, + ): void { + if ($observedSnapshot || $event->connectionName !== $reader + || $event->connection->transactionLevel() === 0 + || ! str_contains(strtolower($event->sql), 'exists') + || ! str_contains($event->sql, 'activity_attempts')) { + return; + } + $observedSnapshot = true; + // The real final handler has established its RR snapshot but + // has not acquired the parent-run lock. Commit its sibling on + // a separate connection before that handler can continue. + config([ + 'workflows.storage.connection' => $writer, + ]); + try { + $this->assertTrue($activities->complete($first, 'concurrent sibling')['recorded']); + $this->assertSame(0, $this->openParentWorkflowTaskCount($run)); + $this->assertSame($size - 1, $run->historyEvents() + ->where('event_type', HistoryEventType::ActivityCompleted->value)->count()); + } finally { + config([ + 'workflows.storage.connection' => $readStorage, + ]); + } + // Prove that merely reloading an ordinary relation would + // still miss the committed sibling on this connection. + $this->assertSame($size - 2, $event->connection->table('workflow_history_events') + ->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::ActivityCompleted->value)->count()); + }); + + $this->assertTrue($activities->complete($last, 'last sibling')['recorded']); + $this->assertTrue($observedSnapshot); + $this->assertSame($size, $run->historyEvents() + ->where('event_type', HistoryEventType::ActivityCompleted->value)->count()); + $this->assertSame(1, $this->openParentWorkflowTaskCount($run)); + $this->assertFalse($activities->complete($last, 'duplicate')['recorded']); + $this->assertSame(1, $this->openParentWorkflowTaskCount($run)); + } finally { + config([ + 'workflows.storage.connection' => $originalStorage, + ]); + DB::purge($writer); + if ($named) { + DB::purge($reader); + } + } + } + + /** + * @return array{WorkflowRun, list} + */ + private function stageActivityGroup(int $size, bool $nested): array + { + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $commands = []; + for ($index = 0; $index < $size; $index++) { + $path = [$this->parallelEntry('parallel-activities:1:' . $size, 'activity', 1, $size, $index)]; + if ($nested && $index >= $size - 2) { + $path[] = $this->parallelEntry( + 'parallel-activities:' . ($size - 1) . ':2', + 'activity', + $size - 1, + 2, + $index - ($size - 2) + ); + } + $commands[] = [ + 'type' => 'schedule_activity', + 'activity_type' => 'test-activity-history', + 'arguments' => Serializer::serialize([]), + ...$path[array_key_last($path)], + 'parallel_group_path' => $path, + ]; + } + $reply = $this->bridge->complete($task->id, $commands); + $this->assertTrue($reply['completed']); + $this->assertCount($size, $reply['created_task_ids']); + $attempts = []; + $activities = $this->app->make(ActivityTaskBridge::class); + foreach ($reply['created_task_ids'] as $id) { + $claim = $activities->claimStatus($id, 'activity-history-' . $id); + $this->assertTrue($claim['claimed']); + $attempts[] = $claim['activity_attempt_id']; + } + + return [$run, $attempts]; + } + /** * @param list>> $parallelPaths * @param list> $commandOverrides