Skip to content
Merged
240 changes: 120 additions & 120 deletions .github/feature-test-timings.json

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion composer.json
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@
"dev-main": "2.0.x-dev"
},
"durable-workflow": {
"product-train": "2.20.3",
"product-train": "2.20.4",
"laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
"laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
},
Expand Down
6 changes: 5 additions & 1 deletion src/V2/Jobs/RunWorkflowTask.php
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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;
Expand Down
20 changes: 16 additions & 4 deletions src/V2/Support/ContinuedRunMessageHandoff.php
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
20 changes: 16 additions & 4 deletions src/V2/Support/PendingSignalCloser.php
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
60 changes: 60 additions & 0 deletions src/V2/Support/WorkflowSignalReadScope.php
Original file line number Diff line number Diff line change
@@ -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;
}
}
Loading
Loading