Keep parent-history row locks within their run

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).

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<int, WorkflowHistoryEvent>
+     */
+    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<string, mixed> $payload
      * @return array{

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();