Avoid cross-run signal gap deadlocks in embedded workflow tasks

src/V2/Jobs/RunWorkflowTask.php

@@ -29,6 +29,7 @@
 use Workflow\V2\Support\WorkerCompatibilityFleet;
 use Workflow\V2\Support\WorkflowContinuationCache;
 use Workflow\V2\Support\WorkflowExecutor;
+use Workflow\V2\Support\WorkflowSignalReadScope;
 use Workflow\V2\Support\WorkflowTaskLease;
 
 final class RunWorkflowTask implements ShouldQueue
@@ -130,7 +131,10 @@ private function executeClaim(WorkflowExecutor $executor, int $claimedAttempt):
                     return null;
                 }
 
-                $nextTask = $executor->run($run, $task, $continuation);
+                $nextTask = WorkflowSignalReadScope::forLockedRun(
+                    $run,
+                    static fn (): ?WorkflowTask => $executor->run($run, $task, $continuation),
+                );
                 $continuation?->offer($task);
 
                 return $nextTask;

src/V2/Support/ContinuedRunMessageHandoff.php

@@ -47,13 +47,25 @@ public static function closePredecessor(
 
     public static function transferInstanceScoped(WorkflowRun $closingRun, WorkflowRun $continuedRun): void
     {
-        $signals = ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
+        $signalQuery = ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
             ->where('workflow_run_id', $closingRun->id)
             ->where('target_scope', 'instance')
             ->where('status', SignalStatus::Received->value)
-            ->whereNull('closed_at')
-            ->lockForUpdate()
-            ->get();
+            ->whereNull('closed_at');
+        if (WorkflowSignalReadScope::allowsConsistentRead($closingRun, $signalQuery->getModel()->getConnection())) {
+            // Lock exact candidates without reserving neighboring index gaps.
+            $signals = $signalQuery->get()
+                ->map(static function (WorkflowSignal $candidate) use ($signalQuery): ?WorkflowSignal {
+                    return (clone $signalQuery)->whereKey($candidate->getKey())
+                        ->lockForUpdate()
+                        ->first();
+                })->filter();
+        } else {
+            // Outside the owned task, the caller may already have an older read view.
+            // Preserve current reads for every other execution path.
+            $signals = $signalQuery->lockForUpdate()
+                ->get();
+        }
         if ($signals->isNotEmpty()) {
             $declaredSignals = RunCommandContract::forRun($closingRun)['signals'];
             $signals = $signals->filter(

src/V2/Support/PendingSignalCloser.php

@@ -30,15 +30,27 @@ public static function closeForTerminalEvent(
             return;
         }
 
-        $signals = ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
+        $signalQuery = ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
             ->where('workflow_run_id', $run->id)
             ->where('status', SignalStatus::Received->value)
             ->orderByRaw('CASE WHEN command_sequence IS NULL THEN 1 ELSE 0 END')
             ->orderBy('command_sequence')
             ->orderBy('received_at')
-            ->orderBy('id')
-            ->lockForUpdate()
-            ->get();
+            ->orderBy('id');
+        if (WorkflowSignalReadScope::allowsConsistentRead($run, $signalQuery->getModel()->getConnection())) {
+            // Lock exact candidates without reserving neighboring index gaps.
+            $signals = $signalQuery->get()
+                ->map(static function (WorkflowSignal $candidate) use ($signalQuery): ?WorkflowSignal {
+                    return (clone $signalQuery)->whereKey($candidate->getKey())
+                        ->lockForUpdate()
+                        ->first();
+                })->filter();
+        } else {
+            // Outside the owned task, the caller may already have an older read view.
+            // Preserve current reads for every other execution path.
+            $signals = $signalQuery->lockForUpdate()
+                ->get();
+        }
 
         foreach ($signals as $signal) {
             if (! $signal instanceof WorkflowSignal) {

src/V2/Support/WorkflowSignalReadScope.php

@@ -0,0 +1,60 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Support;
+
+use Closure;
+use Illuminate\Database\Connection;
+use PDO;
+use WeakMap;
+use Workflow\V2\Models\WorkflowRun;
+
+/** @internal The embedded task has locked this run before its first consistent read. */
+final class WorkflowSignalReadScope
+{
+    /**
+     * @var WeakMap<Connection, array{pdo: PDO, run_id: string}>|null
+     */
+    private static ?WeakMap $scopes = null;
+
+    /**
+     * @template T
+     *
+     * @param  Closure(): T  $operation
+     * @return T
+     */
+    public static function forLockedRun(WorkflowRun $run, Closure $operation): mixed
+    {
+        $connection = $run->getConnection();
+        if ($connection->getDriverName() !== 'mysql' || $connection->transactionLevel() !== 1) {
+            return $operation();
+        }
+
+        self::$scopes ??= new WeakMap();
+        $previous = self::$scopes[$connection] ?? null;
+        self::$scopes[$connection] = [
+            'pdo' => $connection->getPdo(),
+            'run_id' => $run->id,
+        ];
+        try {
+            return $operation();
+        } finally {
+            if ($previous === null) {
+                unset(self::$scopes[$connection]);
+            } else {
+                self::$scopes[$connection] = $previous;
+            }
+        }
+    }
+
+    public static function allowsConsistentRead(WorkflowRun $run, Connection $connection): bool
+    {
+        $scope = self::$scopes[$connection] ?? null;
+
+        return $scope !== null
+            && $connection->transactionLevel() > 0
+            && $scope['pdo'] === $connection->getRawPdo()
+            && $scope['run_id'] === $run->id;
+    }
+}