diff --git a/CHANGELOG.md b/CHANGELOG.md index 642288c7d..c9b3487b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +- Concurrent child completion keeps parent-history row locks on the existing + per-run index on MySQL and MariaDB, avoiding an unrelated sibling history + row in the parent barrier lock cycle. No new index or retry is introduced. + - Require patched Laravel 12.72 or 13.41 minimums for the current Composer security advisories. Keep existing application audits enabled and update application lockfiles before upgrading Workflow (#1133). diff --git a/src/V2/Support/ParallelChildGroup.php b/src/V2/Support/ParallelChildGroup.php index ad1a47b93..ead6f5a56 100644 --- a/src/V2/Support/ParallelChildGroup.php +++ b/src/V2/Support/ParallelChildGroup.php @@ -4,6 +4,8 @@ namespace Workflow\V2\Support; +use Illuminate\Database\Eloquent\Collection; +use Illuminate\Database\Schema\Blueprint; use Workflow\V2\Enums\ActivityStatus; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; @@ -732,10 +734,10 @@ private static function persistSelectionWinner( $events = $parentRun->historyEvents(); if ($lockHistoryForUpdate) { - $events->lockForUpdate(); + $parentRun->setRelation('historyEvents', self::lockedParentHistory($parentRun)); + } else { + $parentRun->setRelation('historyEvents', $events->get()); } - - $parentRun->setRelation('historyEvents', $events->get()); $groupId = $metadata['parallel_group_id']; $memberBase = self::intValue($metadata['selection_member_base_sequence'] ?? null) ?? ($metadata['parallel_group_base_sequence'] + $metadata['parallel_group_index']); @@ -800,7 +802,7 @@ private static function groupCompletedSuccessfully( if ($lockHistoryForUpdate) { // Observe the resolution event committed by the previous holder of // the parent lock even when this transaction has an older snapshot. - $parentRun->setRelation('historyEvents', $parentRun->historyEvents() ->lockForUpdate() ->get()); + $parentRun->setRelation('historyEvents', self::lockedParentHistory($parentRun)); } $activitiesBySequence = collect(RunActivityView::activitiesForRun($parentRun, decodePayloads: false)) @@ -876,6 +878,27 @@ private static function groupCompletedSuccessfully( return true; } + /** + * @return Collection + */ + private static function lockedParentHistory(WorkflowRun $parentRun): Collection + { + $events = $parentRun->historyEvents(); + $model = $events->getRelated(); + $connection = $model->getConnection(); + if (in_array($connection->getDriverName(), ['mysql', 'mariadb'], true)) { + // A table scan can lock an uncommitted sibling event while that + // sibling waits for our parent-run lock. Use the migration's + // per-run index, including Laravel's configured table/index prefix. + $connection->getSchemaBuilder(); + $index = (new Blueprint($connection, $model->getTable())) + ->unique(['workflow_run_id', 'sequence'])->index; + $events->forceIndex($index); + } + return $events->lockForUpdate() + ->get(); + } + /** * @param array $payload * @return array{ diff --git a/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php b/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php index 29e20aaf6..36cfae242 100644 --- a/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php +++ b/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php @@ -299,12 +299,50 @@ public function testConcurrentChildCompletionsSerializeOneBarrierWakeOnParentRun $children = []; $results = []; + $completionGates = []; + $parentRunId = $parentRun->id; DB::purge(); try { foreach ($childTasks as $childTask) { - $children[] = $this->forkDatabaseOperation(static function () use ($childTask): array { + $gate = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + $this->assertIsArray($gate); + stream_set_timeout($gate[0], 15); + stream_set_timeout($gate[1], 15); + $completionGates[] = $gate[0]; + $children[] = $this->forkDatabaseOperation(static function () use ( + $childTask, + $parentRunId, + $gate, + ): array { + fclose($gate[0]); + $appended = false; + $parentLocked = false; + DB::listen(static function (QueryExecuted $query) use ( + $childTask, + $parentRunId, + $gate, + &$appended, + &$parentLocked, + ): void { + if (! $appended && str_starts_with(strtolower($query->sql), 'insert') + && str_contains($query->sql, 'workflow_history_events') + && in_array($childTask->workflow_run_id, $query->bindings, true) + && in_array(HistoryEventType::WorkflowCompleted->value, $query->bindings, true)) { + $appended = true; + fwrite($gate[1], "appended\n"); + if (fgets($gate[1]) !== "continue\n") { + throw new RuntimeException('Child completion gate was not released.'); + } + } + if ($appended && ! $parentLocked && str_contains($query->sql, 'workflow_runs') + && str_contains(strtolower($query->sql), 'for update') + && in_array($parentRunId, $query->bindings, true)) { + $parentLocked = true; + fwrite($gate[1], "parent-locked\n"); + } + }); /** @var WorkflowTaskBridge $bridge */ $bridge = app(WorkflowTaskBridge::class); @@ -315,6 +353,7 @@ public function testConcurrentChildCompletionsSerializeOneBarrierWakeOnParentRun ]), ]]); }); + fclose($gate[1]); } foreach ($children as $child) { @@ -325,12 +364,24 @@ public function testConcurrentChildCompletionsSerializeOneBarrierWakeOnParentRun $this->releaseDatabaseOperation($child); } + // Keep both child history writes uncommitted. The first child + // then owns the parent while the sibling still holds its row. + foreach ($completionGates as $gate) { + $this->assertSame("appended\n", fgets($gate)); + } + fwrite($completionGates[0], "continue\n"); + $this->assertSame("parent-locked\n", fgets($completionGates[0])); + fwrite($completionGates[1], "continue\n"); + $results = array_map(fn (array $child): array => $this->awaitDatabaseOperation($child), $children); $children = []; } finally { foreach ($children as $child) { $this->terminateDatabaseOperation($child); } + foreach ($completionGates as $gate) { + fclose($gate); + } } DB::reconnect();