From e7d18f719bf91bf8da677ba8c81eb3a9a3f6ab4a Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 24 Oct 2026 03:42:19 +0000 Subject: [PATCH 1/6] test: reproduce retention against expired activity outcomes --- .../ActivityOutcomeRetentionProcessTest.php | 87 ++++++++ .../Unit/V2/ActivityOutcomeRetentionTest.php | 191 ++++++++++++++++++ 2 files changed, 278 insertions(+) create mode 100644 tests/Support/ActivityOutcomeRetentionProcessTest.php create mode 100644 tests/Unit/V2/ActivityOutcomeRetentionTest.php diff --git a/tests/Support/ActivityOutcomeRetentionProcessTest.php b/tests/Support/ActivityOutcomeRetentionProcessTest.php new file mode 100644 index 000000000..8751b94cb --- /dev/null +++ b/tests/Support/ActivityOutcomeRetentionProcessTest.php @@ -0,0 +1,87 @@ +input(); + Date::setTestNow($input['observed_at']); + $connection = DB::connection(); + if ($input['outcome'] === 'pruned') { + $connection->listen(static function (QueryExecuted $event) use ($input): void { + if (! str_contains(strtolower($event->sql), 'for update') + || ! str_contains($event->sql, 'workflow_runs')) { + return; + } + file_put_contents($input['ready'], 'run locked'); + while (! is_file($input['continue'])) { + if (hrtime(true) >= $input['deadline']) { + self::fail('Retention exceeded the original process budget.'); + } + usleep(1_000); + } + }); + $report = WorkflowRunRetentionCleanup::pruneRun($input['run_id']); + self::assertGreaterThan(0, $report['history_events_deleted']); + $reply = ['recorded' => false, 'reason' => 'pruned']; + } else { + // Signal before this query can wait on retention's run lock. The + // preceding implementation already holds task/execution locks here. + $connection->beforeExecuting(static function (string $query) use ($input): void { + if (str_contains(strtolower($query), 'for update') + && str_contains($query, 'workflow_runs')) { + file_put_contents($input['ready'], 'requesting run lock'); + } + }); + $bridge = $this->app->make(ActivityTaskBridge::class); + $reply = $input['outcome'] === 'completed' + ? $bridge->complete($input['attempt_id'], 'discarded late result') + : $bridge->fail($input['attempt_id'], ['message' => 'discarded late failure']); + self::assertFalse($reply['recorded']); + } + + file_put_contents($input['report'], json_encode([ + 'pid' => getmypid(), + 'recorded' => $reply['recorded'], + 'reason' => $reply['reason'], + ], JSON_THROW_ON_ERROR)); + } + + protected function getPackageProviders($app): array + { + return [WorkflowServiceProvider::class]; + } + + protected function defineEnvironment($app): void + { + $input = $this->input(); + $app['config']->set('database.default', 'retention_process'); + $app['config']->set('database.connections.retention_process', $input['database']); + $app['config']->set('workflows.storage.connection', 'retention_process'); + $app['config']->set('queue.default', 'sync'); + $app['config']->set('cache.default', 'array'); + $app['config']->set('logging.default', 'null'); + } + + private function input(): array + { + $path = getenv('DWF_RETENTION_PROCESS_INPUT'); + if (! is_string($path) || $path === '') { + $this->markTestSkipped('Invoked only by the concurrent retention control.'); + } + + return json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + } +} diff --git a/tests/Unit/V2/ActivityOutcomeRetentionTest.php b/tests/Unit/V2/ActivityOutcomeRetentionTest.php new file mode 100644 index 000000000..094cb6478 --- /dev/null +++ b/tests/Unit/V2/ActivityOutcomeRetentionTest.php @@ -0,0 +1,191 @@ +freezeTime(); + config(['workflows.v2.task_dispatch_mode' => 'poll']); + } + + public static function retainedOutcomes(): array + { + $cases = []; + foreach ([RunStatus::Completed, RunStatus::Failed, RunStatus::Cancelled, RunStatus::Terminated] as $status) { + foreach ([false, true] as $reclaimed) { + foreach (['completed', 'failed'] as $outcome) { + $cases[$status->value . '/' . (int) $reclaimed . '/' . $outcome] = [$status, $reclaimed, $outcome]; + } + } + } + + return $cases; + } + + #[DataProvider('retainedOutcomes')] + public function testPruningAndRefusedOutcomeCannotDeadlockOrRestoreDetails( + RunStatus $status, + bool $reclaimed, + string $outcome, + ): void { + if (DB::connection()->getDriverName() === 'sqlite') { + $this->markTestSkipped('This deterministic row-lock interleaving requires PostgreSQL or MySQL.'); + } + $deadline = hrtime(true) + 30_000_000_000; + [, $claim, $run] = $this->claimedOriginal(); + $this->travel(301)->seconds(); + if ($reclaimed) { + self::assertTrue(ActivityTimeoutEnforcer::enforceLeaseExpiry($claim['task_id'])['enforced']); + } + $run->forceFill(['status' => $status, 'closed_at' => now()])->save(); + $reports = $this->interleavePruning($run, $claim, $outcome, $deadline); + + self::assertNotSame($reports[0]['pid'], $reports[1]['pid']); + self::assertSame('pruned', $reports[0]['reason']); + self::assertFalse($reports[1]['recorded']); + self::assertContains($reports[1]['reason'], [ + 'stale_attempt', 'attempt_not_found', 'task_not_found', 'activity_execution_missing', + ]); + self::assertSame($status, $run->fresh()->status); + self::assertNotNull($run->fresh()->details_pruned_at); + self::assertSame(0, $run->historyEvents()->count()); + foreach ([ActivityAttempt::class, ActivityExecution::class, WorkflowTask::class, WorkflowTimelineEntry::class] as $model) { + self::assertSame(0, $model::query()->where('workflow_run_id', $run->id)->count()); + } + self::assertNull(WorkflowRunSummary::query()->find($run->id)); + } + + public function testRefusedExpiredOutcomeDoesNotPreparePayloadOrWriteHistory(): void + { + [$bridge, $claim, $run] = $this->claimedOriginal(); + $this->travel(301)->seconds(); + $before = $run->historyEvents()->get()->map->getAttributes()->all(); + $reply = $bridge->complete($claim['activity_attempt_id'], new class() { + public function __serialize(): array + { + throw new \RuntimeException('An expired result must not be serialized.'); + } + }); + + self::assertFalse($reply['recorded']); + self::assertSame('lease_expired', $reply['reason']); + self::assertSame($before, $run->historyEvents()->get()->map->getAttributes()->all()); + self::assertSame(ActivityAttemptStatus::Running, ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status); + } + + private function claimedOriginal(): array + { + $workflow = WorkflowStub::make(TestRetryWorkflow::class, 'outcome-retention'); + $workflow->start('Ada'); + $task = WorkflowTask::query()->where('task_type', TaskType::Workflow)->where('status', TaskStatus::Ready)->sole(); + $this->app->call([new RunWorkflowTask($task->id), 'handle']); + $bridge = $this->app->make(ActivityTaskBridge::class); + $task = $bridge->poll(null, null, 1)[0]; + $claim = $bridge->claimStatus($task['task_id'], 'original-worker'); + self::assertTrue($claim['claimed']); + + return [$bridge, $claim, WorkflowRun::query()->findOrFail($workflow->runId())]; + } + + private function interleavePruning(WorkflowRun $run, array $claim, string $outcome, int $deadline): array + { + $directory = sys_get_temp_dir() . '/workflow-outcome-retention-' . bin2hex(random_bytes(8)); + mkdir($directory, 0700); + $workers = []; + try { + foreach (['pruned', $outcome] as $index => $operation) { + $path = $directory . '/input-' . $index . '.json'; + file_put_contents($path, json_encode([ + 'database' => DB::connection($run->getConnectionName())->getConfig(), + 'run_id' => $run->id, + 'attempt_id' => $claim['activity_attempt_id'], + 'outcome' => $operation, + 'observed_at' => now()->toISOString(), + 'deadline' => $deadline, + 'ready' => $directory . '/ready-' . $index, + 'continue' => $directory . '/continue', + 'report' => $directory . '/report-' . $index . '.json', + ], JSON_THROW_ON_ERROR)); + $worker = new Process([ + PHP_BINARY, + dirname(__DIR__, 3) . '/vendor/bin/phpunit', + dirname(__DIR__, 2) . '/Support/ActivityOutcomeRetentionProcessTest.php', + '--do-not-cache-result', + ], dirname(__DIR__, 3), [ + 'DWF_RETENTION_PROCESS_INPUT' => $path, + 'APP_SERVICES_CACHE' => $directory . '/services-' . $index . '.php', + 'APP_PACKAGES_CACHE' => $directory . '/packages-' . $index . '.php', + 'XDEBUG_MODE' => 'off', + ], timeout: 30); + $workers[] = $worker; + $worker->start(); + $this->awaitFile($directory . '/ready-' . $index, $workers, $deadline); + } + file_put_contents($directory . '/continue', 'continue'); + while ($workers[0]->isRunning() || $workers[1]->isRunning()) { + if (hrtime(true) >= $deadline) { + self::fail('The original concurrent retention budget expired.'); + } + usleep(1_000); + } + foreach ($workers as $worker) { + self::assertSame(0, $worker->getExitCode(), substr($worker->getErrorOutput() . $worker->getOutput(), -8_192)); + } + + return [ + json_decode(file_get_contents($directory . '/report-0.json'), true, flags: JSON_THROW_ON_ERROR), + json_decode(file_get_contents($directory . '/report-1.json'), true, flags: JSON_THROW_ON_ERROR), + ]; + } finally { + foreach ($workers as $worker) { + $worker->stop(2); + } + foreach (glob($directory . '/*') ?: [] as $path) { + unlink($path); + } + rmdir($directory); + } + } + + private function awaitFile(string $path, array $workers, int $deadline): void + { + while (! is_file($path)) { + foreach ($workers as $worker) { + if (! $worker->isRunning()) { + self::fail(substr($worker->getErrorOutput() . $worker->getOutput(), -8_192)); + } + } + if (hrtime(true) >= $deadline) { + self::fail('The original retention interleaving budget expired.'); + } + usleep(1_000); + } + } +} From ecf2b3d6bf039aa60da706489cfcbe086d36c065 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 24 Oct 2026 03:46:17 +0000 Subject: [PATCH 2/6] test: cover inline outcomes in the retention lock interleaving --- .../ActivityOutcomeRetentionProcessTest.php | 22 +++++++++++++++---- .../Unit/V2/ActivityOutcomeRetentionTest.php | 12 +++++++--- 2 files changed, 27 insertions(+), 7 deletions(-) diff --git a/tests/Support/ActivityOutcomeRetentionProcessTest.php b/tests/Support/ActivityOutcomeRetentionProcessTest.php index 8751b94cb..1ebe8a3fd 100644 --- a/tests/Support/ActivityOutcomeRetentionProcessTest.php +++ b/tests/Support/ActivityOutcomeRetentionProcessTest.php @@ -8,9 +8,11 @@ use Illuminate\Support\Facades\Date; use Illuminate\Support\Facades\DB; use Orchestra\Testbench\TestCase; +use RuntimeException; use Workflow\Providers\WorkflowServiceProvider; use Workflow\V2\Contracts\ActivityTaskBridge; use Workflow\V2\Support\WorkflowRunRetentionCleanup; +use Workflow\V2\Support\ActivityOutcomeRecorder; final class ActivityOutcomeRetentionProcessTest extends TestCase { @@ -45,10 +47,22 @@ public function testOutcomeAndRetention(): void file_put_contents($input['ready'], 'requesting run lock'); } }); - $bridge = $this->app->make(ActivityTaskBridge::class); - $reply = $input['outcome'] === 'completed' - ? $bridge->complete($input['attempt_id'], 'discarded late result') - : $bridge->fail($input['attempt_id'], ['message' => 'discarded late failure']); + if ($input['direct']) { + $reply = ActivityOutcomeRecorder::record( + $input['task_id'], + $input['attempt_id'], + $input['attempt_number'], + 'discarded late result', + $input['outcome'] === 'failed' ? new RuntimeException('discarded late failure') : null, + 2, + 5, + ); + } else { + $bridge = $this->app->make(ActivityTaskBridge::class); + $reply = $input['outcome'] === 'completed' + ? $bridge->complete($input['attempt_id'], 'discarded late result') + : $bridge->fail($input['attempt_id'], ['message' => 'discarded late failure']); + } self::assertFalse($reply['recorded']); } diff --git a/tests/Unit/V2/ActivityOutcomeRetentionTest.php b/tests/Unit/V2/ActivityOutcomeRetentionTest.php index 094cb6478..b3008fafa 100644 --- a/tests/Unit/V2/ActivityOutcomeRetentionTest.php +++ b/tests/Unit/V2/ActivityOutcomeRetentionTest.php @@ -41,7 +41,9 @@ public static function retainedOutcomes(): array foreach ([RunStatus::Completed, RunStatus::Failed, RunStatus::Cancelled, RunStatus::Terminated] as $status) { foreach ([false, true] as $reclaimed) { foreach (['completed', 'failed'] as $outcome) { - $cases[$status->value . '/' . (int) $reclaimed . '/' . $outcome] = [$status, $reclaimed, $outcome]; + $key = $status->value . '/' . (int) $reclaimed . '/' . $outcome; + $cases[$key . '/bridge'] = [$status, $reclaimed, $outcome, false]; + $cases[$key . '/inline'] = [$status, $reclaimed, $outcome, true]; } } } @@ -54,6 +56,7 @@ public function testPruningAndRefusedOutcomeCannotDeadlockOrRestoreDetails( RunStatus $status, bool $reclaimed, string $outcome, + bool $direct, ): void { if (DB::connection()->getDriverName() === 'sqlite') { $this->markTestSkipped('This deterministic row-lock interleaving requires PostgreSQL or MySQL.'); @@ -65,7 +68,7 @@ public function testPruningAndRefusedOutcomeCannotDeadlockOrRestoreDetails( self::assertTrue(ActivityTimeoutEnforcer::enforceLeaseExpiry($claim['task_id'])['enforced']); } $run->forceFill(['status' => $status, 'closed_at' => now()])->save(); - $reports = $this->interleavePruning($run, $claim, $outcome, $deadline); + $reports = $this->interleavePruning($run, $claim, $outcome, $deadline, $direct); self::assertNotSame($reports[0]['pid'], $reports[1]['pid']); self::assertSame('pruned', $reports[0]['reason']); @@ -114,7 +117,7 @@ private function claimedOriginal(): array return [$bridge, $claim, WorkflowRun::query()->findOrFail($workflow->runId())]; } - private function interleavePruning(WorkflowRun $run, array $claim, string $outcome, int $deadline): array + private function interleavePruning(WorkflowRun $run, array $claim, string $outcome, int $deadline, bool $direct): array { $directory = sys_get_temp_dir() . '/workflow-outcome-retention-' . bin2hex(random_bytes(8)); mkdir($directory, 0700); @@ -126,6 +129,9 @@ private function interleavePruning(WorkflowRun $run, array $claim, string $outco 'database' => DB::connection($run->getConnectionName())->getConfig(), 'run_id' => $run->id, 'attempt_id' => $claim['activity_attempt_id'], + 'task_id' => $claim['task_id'], + 'attempt_number' => $claim['attempt_number'], + 'direct' => $direct, 'outcome' => $operation, 'observed_at' => now()->toISOString(), 'deadline' => $deadline, From dec2aff3e1ec6baf4d95b6f6303c24efa72ac30e Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 24 Oct 2026 03:51:54 +0000 Subject: [PATCH 3/6] Fence lost activity outcomes before retention can delete details --- CHANGELOG.md | 10 ++ composer.json | 2 +- docs/activity-outcome-retention.md | 33 ++++ src/V2/Support/ActivityOutcomeRecorder.php | 93 +++++++++++ .../Support/WorkflowRunRetentionCleanup.php | 13 +- .../ActivityOutcomeRetentionProcessTest.php | 11 +- .../Unit/V2/ActivityOutcomeRetentionTest.php | 158 ++++++++++++++++-- 7 files changed, 302 insertions(+), 18 deletions(-) create mode 100644 docs/activity-outcome-retention.md diff --git a/CHANGELOG.md b/CHANGELOG.md index a4acba4b4..a8943dd48 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## 2.12.2 - Unreleased + +- Refuse expired or settled activity callbacks under the original run lock + before acquiring task or execution locks. Preserve committed renewals and + the normal live-outcome path. Concurrent retention can finish without a + reverse lock dependency or restored details. +- Acquire SQLite writer admission before reading a retained run for pruning, + within the existing transaction policy and original publication budget. + This correction emits no new history event and needs no new event reader. + ## 2.12.1 - Unreleased - Require the upstream TokenGuard string-validation fix on Laravel 11, raising diff --git a/composer.json b/composer.json index 618cd1637..81915cc0a 100644 --- a/composer.json +++ b/composer.json @@ -87,7 +87,7 @@ "dev-main": "2.0.x-dev" }, "durable-workflow": { - "product-train": "2.12.1", + "product-train": "2.12.2", "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json", "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json" }, diff --git a/docs/activity-outcome-retention.md b/docs/activity-outcome-retention.md new file mode 100644 index 000000000..cd7994754 --- /dev/null +++ b/docs/activity-outcome-retention.md @@ -0,0 +1,33 @@ +# Activity outcome refusals and retention + +A late activity callback and pruning can reach the same terminal run at the +same time. Pruning locks the run before deleting detail rows. A callback must +not hold task or execution locks while waiting for that run lock if its attempt +has already lost authority. + +The prospective Core 2.12.2 correction first checks an expired or settled +attempt under the run lock, using fresh stored task, execution and attempt +bindings. If pruning has completed, it refuses the callback without recreating +rows. If a renewal committed before the lock, its current lease remains +authoritative and normal outcome recording performs its existing checks. +Ordinary live attempts keep their existing outcome path and deadlines. + +This refusal does not serialize a result, resolve an external payload, record a +new diagnostic, alter the retry policy or grant business-outcome authority. It +is independent of the proposed rejected-outcome audit and its reader rollout. +The canonical timeout and cancellation mechanisms remain responsible for +settling the original attempt. + +SQLite pruning acquires the writer before reading its conditional snapshot. +Admission does not change the run's identity or ORM timestamps, and already +claimed outcomes retain their original monotonic publication deadline. + +The regression controls use two actual database processes. Retention holds the +run lock before the callback requests it, then deletes the same original +details. Both inline and service-bridge paths must finish without a deadlock or +restored detail, across completed, failed, cancelled and terminated runs. Other +controls cover committed renewal, result-preparation refusal and SQLite writer +ownership before reading the retention snapshot. + +Source qualification is not publication. Normal package installation and any +affected Server image adoption require their own exact-artifact acceptance. diff --git a/src/V2/Support/ActivityOutcomeRecorder.php b/src/V2/Support/ActivityOutcomeRecorder.php index 275419f4e..22fb56084 100644 --- a/src/V2/Support/ActivityOutcomeRecorder.php +++ b/src/V2/Support/ActivityOutcomeRecorder.php @@ -41,6 +41,11 @@ public static function record( ?string $codec = null, ?int $admissionDeadline = null, ): array { + $refused = self::refusedExpiredAttempt($taskId, $attemptId, $attemptCount, $admissionDeadline); + if ($refused !== null) { + return $refused; + } + return SqliteTransaction::forTask($taskId, static function () use ( $taskId, $attemptId, @@ -508,6 +513,94 @@ public static function recordForAttempt( ); } + /** + * Fence an already lost outcome without taking task/execution locks in the + * reverse order to retention. A normal live outcome keeps its existing path. + * + * @return array{recorded: bool, reason: string|null, next_task: WorkflowTask|null}|null + */ + private static function refusedExpiredAttempt( + string $taskId, + string $attemptId, + int $attemptNumber, + ?int $admissionDeadline, + ): ?array { + $original = ActivityAttempt::query()->find($attemptId); + if (! $original instanceof ActivityAttempt || $original->id !== $attemptId + || $original->workflow_task_id !== $taskId || $original->attempt_number !== $attemptNumber) { + return null; + } + if ($original->status === ActivityAttemptStatus::Running + && $original->lease_expires_at !== null && now() + ->lt($original->lease_expires_at)) { + return null; + } + + $connection = $original->getConnection(); + $runId = (string) $original->workflow_run_id; + + return SqliteTransaction::run(static function () use ($original): ?array { + return self::refusedExpiredAttemptUnderRunLock($original); + }, admission: static function () use ($connection, $runId): void { + WorkflowRun::on($connection->getName())->whereKey($runId)->toBase()->update([ + 'id' => $runId, + ]); + }, admissionDeadline: $admissionDeadline, connection: $connection); + } + + /** + * @return array{recorded: bool, reason: string|null, next_task: WorkflowTask|null}|null + */ + private static function refusedExpiredAttemptUnderRunLock(ActivityAttempt $original): ?array + { + $connection = $original->getConnectionName(); + $run = WorkflowRun::on($connection)->whereKey($original->workflow_run_id)->lockForUpdate()->first(); + if (! $run instanceof WorkflowRun || $run->details_pruned_at !== null) { + return self::ignored('stale_attempt'); + } + $attempt = ActivityAttempt::on($connection)->find($original->id); + $task = WorkflowTask::on($connection)->find($original->workflow_task_id); + $execution = ActivityExecution::on($connection)->find($original->activity_execution_id); + if (! $attempt instanceof ActivityAttempt || ! $task instanceof WorkflowTask + || ! $execution instanceof ActivityExecution + || $attempt->id !== $original->id || $attempt->attempt_number !== $original->attempt_number + || $attempt->workflow_task_id !== $task->id || $attempt->activity_execution_id !== $execution->id + || $attempt->workflow_run_id !== $run->id || $execution->workflow_run_id !== $run->id + || $task->workflow_run_id !== $run->id || $task->namespace !== $run->namespace + || ($task->payload['activity_execution_id'] ?? null) !== $execution->id) { + return self::ignored('stale_attempt'); + } + + $reason = self::expiredAttemptReason($run, $execution, $task, $attempt); + + return $reason === null ? null : self::ignored($reason); + } + + private static function expiredAttemptReason( + WorkflowRun $run, + ActivityExecution $execution, + WorkflowTask $task, + ActivityAttempt $attempt, + ): ?string { + if (in_array($run->status, [RunStatus::Cancelled, RunStatus::Terminated], true) + && ! ActivityAbandonment::allows($run, $execution)) { + return $run->status === RunStatus::Terminated ? 'run_terminated' : 'run_cancelled'; + } + if ($attempt->status !== ActivityAttemptStatus::Running || $task->status !== TaskStatus::Leased + || $task->attempt_count !== $attempt->attempt_number + || $execution->attempt_count !== $attempt->attempt_number + || $execution->current_attempt_id !== $attempt->id) { + return 'stale_attempt'; + } + if (ActivityTimeoutEnforcer::hasExpiredDeadline($execution, now()) + && ActivityAbandonment::allows($run, $execution)) { + return 'activity_deadline_elapsed'; + } + + // Renewals that committed before the run lock must keep their authority. + return ActivityAttemptAuthority::expiredReason($task, $execution, now(), $run); + } + /** * Close a standalone-activity host run with the activity's terminal * outcome. On success the run is marked Completed and the run output diff --git a/src/V2/Support/WorkflowRunRetentionCleanup.php b/src/V2/Support/WorkflowRunRetentionCleanup.php index f3902ab25..edf93e46c 100644 --- a/src/V2/Support/WorkflowRunRetentionCleanup.php +++ b/src/V2/Support/WorkflowRunRetentionCleanup.php @@ -75,7 +75,7 @@ public static function pruneRun(WorkflowRun|string $run): array ->getModel() ->getConnection(); - return $runConnection->transaction(static function () use ($runId): array { + return SqliteTransaction::run(static function () use ($runId): array { /** @var WorkflowRun $lockedRun */ $lockedRun = ConfiguredV2Models::query('run_model', WorkflowRun::class) ->whereKey($runId) @@ -171,7 +171,16 @@ public static function pruneRun(WorkflowRun|string $run): array } return $report; - }); + }, admission: static function () use ($runId): void { + $runs = ConfiguredV2Models::query('run_model', WorkflowRun::class); + $key = $runs->getModel() + ->getKeyName(); + $runs->whereKey($runId) + ->toBase() + ->update([ + $key => $runId, + ]); + }, connection: $runConnection); } /** diff --git a/tests/Support/ActivityOutcomeRetentionProcessTest.php b/tests/Support/ActivityOutcomeRetentionProcessTest.php index 1ebe8a3fd..26b1ed7e1 100644 --- a/tests/Support/ActivityOutcomeRetentionProcessTest.php +++ b/tests/Support/ActivityOutcomeRetentionProcessTest.php @@ -11,8 +11,8 @@ use RuntimeException; use Workflow\Providers\WorkflowServiceProvider; use Workflow\V2\Contracts\ActivityTaskBridge; -use Workflow\V2\Support\WorkflowRunRetentionCleanup; use Workflow\V2\Support\ActivityOutcomeRecorder; +use Workflow\V2\Support\WorkflowRunRetentionCleanup; final class ActivityOutcomeRetentionProcessTest extends TestCase { @@ -37,7 +37,10 @@ public function testOutcomeAndRetention(): void }); $report = WorkflowRunRetentionCleanup::pruneRun($input['run_id']); self::assertGreaterThan(0, $report['history_events_deleted']); - $reply = ['recorded' => false, 'reason' => 'pruned']; + $reply = [ + 'recorded' => false, + 'reason' => 'pruned', + ]; } else { // Signal before this query can wait on retention's run lock. The // preceding implementation already holds task/execution locks here. @@ -61,7 +64,9 @@ public function testOutcomeAndRetention(): void $bridge = $this->app->make(ActivityTaskBridge::class); $reply = $input['outcome'] === 'completed' ? $bridge->complete($input['attempt_id'], 'discarded late result') - : $bridge->fail($input['attempt_id'], ['message' => 'discarded late failure']); + : $bridge->fail($input['attempt_id'], [ + 'message' => 'discarded late failure', + ]); } self::assertFalse($reply['recorded']); } diff --git a/tests/Unit/V2/ActivityOutcomeRetentionTest.php b/tests/Unit/V2/ActivityOutcomeRetentionTest.php index b3008fafa..caa1e8f17 100644 --- a/tests/Unit/V2/ActivityOutcomeRetentionTest.php +++ b/tests/Unit/V2/ActivityOutcomeRetentionTest.php @@ -6,10 +6,13 @@ use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; +use PDO; +use PDOException; use PHPUnit\Framework\Attributes\DataProvider; use Symfony\Component\Process\Process; use Tests\Fixtures\V2\TestRetryWorkflow; use Tests\TestCase; +use Workflow\V2\Contracts\ActivityLeaseRenewal; use Workflow\V2\Contracts\ActivityTaskBridge; use Workflow\V2\Enums\ActivityAttemptStatus; use Workflow\V2\Enums\RunStatus; @@ -22,7 +25,9 @@ use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Models\WorkflowTimelineEntry; +use Workflow\V2\Support\ActivityOutcomeRecorder; use Workflow\V2\Support\ActivityTimeoutEnforcer; +use Workflow\V2\Support\WorkflowRunRetentionCleanup; use Workflow\V2\WorkflowStub; final class ActivityOutcomeRetentionTest extends TestCase @@ -32,7 +37,9 @@ protected function setUp(): void parent::setUp(); Queue::fake(); $this->freezeTime(); - config(['workflows.v2.task_dispatch_mode' => 'poll']); + config([ + 'workflows.v2.task_dispatch_mode' => 'poll', + ]); } public static function retainedOutcomes(): array @@ -63,11 +70,15 @@ public function testPruningAndRefusedOutcomeCannotDeadlockOrRestoreDetails( } $deadline = hrtime(true) + 30_000_000_000; [, $claim, $run] = $this->claimedOriginal(); - $this->travel(301)->seconds(); + $this->travel(301) + ->seconds(); if ($reclaimed) { self::assertTrue(ActivityTimeoutEnforcer::enforceLeaseExpiry($claim['task_id'])['enforced']); } - $run->forceFill(['status' => $status, 'closed_at' => now()])->save(); + $run->forceFill([ + 'status' => $status, + 'closed_at' => now(), + ])->save(); $reports = $this->interleavePruning($run, $claim, $outcome, $deadline, $direct); self::assertNotSame($reports[0]['pid'], $reports[1]['pid']); @@ -79,7 +90,12 @@ public function testPruningAndRefusedOutcomeCannotDeadlockOrRestoreDetails( self::assertSame($status, $run->fresh()->status); self::assertNotNull($run->fresh()->details_pruned_at); self::assertSame(0, $run->historyEvents()->count()); - foreach ([ActivityAttempt::class, ActivityExecution::class, WorkflowTask::class, WorkflowTimelineEntry::class] as $model) { + foreach ([ + ActivityAttempt::class, + ActivityExecution::class, + WorkflowTask::class, + WorkflowTimelineEntry::class, + ] as $model) { self::assertSame(0, $model::query()->where('workflow_run_id', $run->id)->count()); } self::assertNull(WorkflowRunSummary::query()->find($run->id)); @@ -88,8 +104,12 @@ public function testPruningAndRefusedOutcomeCannotDeadlockOrRestoreDetails( public function testRefusedExpiredOutcomeDoesNotPreparePayloadOrWriteHistory(): void { [$bridge, $claim, $run] = $this->claimedOriginal(); - $this->travel(301)->seconds(); - $before = $run->historyEvents()->get()->map->getAttributes()->all(); + $this->travel(301) + ->seconds(); + $before = $run->historyEvents() + ->get() + ->map->getAttributes() + ->all(); $reply = $bridge->complete($claim['activity_attempt_id'], new class() { public function __serialize(): array { @@ -100,14 +120,118 @@ public function __serialize(): array self::assertFalse($reply['recorded']); self::assertSame('lease_expired', $reply['reason']); self::assertSame($before, $run->historyEvents()->get()->map->getAttributes()->all()); - self::assertSame(ActivityAttemptStatus::Running, ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status); + self::assertSame( + ActivityAttemptStatus::Running, + ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status + ); + } + + public static function outcomeEntryPoints(): array + { + return [ + 'service bridge' => [false], + 'inline callback' => [true], + ]; + } + + #[DataProvider('outcomeEntryPoints')] + public function testACommittedRenewalKeepsAuthorityAfterAnOlderRead(bool $direct): void + { + [$bridge, $claim] = $this->claimedOriginal(); + self::assertInstanceOf(ActivityLeaseRenewal::class, $bridge); + $this->travel(295) + ->seconds(); + $armed = true; + $observedExpiry = null; + ActivityAttempt::retrieved(function (ActivityAttempt $attempt) use ( + &$armed, + &$observedExpiry, + $bridge, + $claim, + ): void { + if (! $armed || $attempt->id !== $claim['activity_attempt_id']) { + return; + } + $armed = false; + $observedExpiry = $attempt->lease_expires_at; + self::assertTrue( + $bridge->renewLease($claim['activity_attempt_id'], $claim['lease_owner'])['lease_renewed'] + ); + $this->travel(10) + ->seconds(); + }); + + $reply = $direct + ? ActivityOutcomeRecorder::record( + $claim['task_id'], + $claim['activity_attempt_id'], + $claim['attempt_number'], + 'accepted', + null, + 2, + 5 + ) + : $bridge->complete($claim['activity_attempt_id'], 'accepted'); + + self::assertFalse($armed); + self::assertNotNull($observedExpiry); + self::assertTrue(now()->gte($observedExpiry)); + self::assertTrue($reply['recorded']); + self::assertSame( + ActivityAttemptStatus::Completed, + ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status + ); + } + + public function testSqliteRetentionOwnsTheWriterBeforeReadingItsRunSnapshot(): void + { + $connection = DB::connection(); + if ($connection->getDriverName() !== 'sqlite' || $connection->getDatabaseName() === ':memory:') { + $this->markTestSkipped('Writer admission requires two connections to the same SQLite file.'); + } + [, , $run] = $this->claimedOriginal(); + $run->forceFill([ + 'status' => RunStatus::Completed, + 'closed_at' => now(), + ])->save(); + $checked = false; + $writerHeld = false; + $database = $connection->getDatabaseName(); + $connection->listen(static function ($event) use (&$checked, &$writerHeld, $database): void { + if ($checked || ! str_starts_with(strtolower(ltrim($event->sql)), 'select') + || ! str_contains($event->sql, 'workflow_runs')) { + return; + } + $checked = true; + $other = new PDO('sqlite:' . $database, options: [ + PDO::ATTR_TIMEOUT => 0, + ]); + $other->exec('PRAGMA busy_timeout=0'); + try { + $other->exec('BEGIN IMMEDIATE'); + $other->exec('ROLLBACK'); + } catch (PDOException $error) { + self::assertContains(((int) ($error->errorInfo[1] ?? 0)) & 255, [5, 6]); + $writerHeld = true; + } + }); + + WorkflowRunRetentionCleanup::pruneRun($run); + + self::assertTrue($checked); + self::assertTrue($writerHeld, 'The retention snapshot must already own SQLite writer admission.'); + self::assertNotNull($run->fresh()->details_pruned_at); + self::assertSame(0, $run->historyEvents()->count()); } private function claimedOriginal(): array { $workflow = WorkflowStub::make(TestRetryWorkflow::class, 'outcome-retention'); $workflow->start('Ada'); - $task = WorkflowTask::query()->where('task_type', TaskType::Workflow)->where('status', TaskStatus::Ready)->sole(); + $task = WorkflowTask::query()->where('task_type', TaskType::Workflow)->where( + 'status', + TaskStatus::Ready + )->sole(); $this->app->call([new RunWorkflowTask($task->id), 'handle']); $bridge = $this->app->make(ActivityTaskBridge::class); $task = $bridge->poll(null, null, 1)[0]; @@ -117,8 +241,13 @@ private function claimedOriginal(): array return [$bridge, $claim, WorkflowRun::query()->findOrFail($workflow->runId())]; } - private function interleavePruning(WorkflowRun $run, array $claim, string $outcome, int $deadline, bool $direct): array - { + private function interleavePruning( + WorkflowRun $run, + array $claim, + string $outcome, + int $deadline, + bool $direct + ): array { $directory = sys_get_temp_dir() . '/workflow-outcome-retention-' . bin2hex(random_bytes(8)); mkdir($directory, 0700); $workers = []; @@ -133,7 +262,8 @@ private function interleavePruning(WorkflowRun $run, array $claim, string $outco 'attempt_number' => $claim['attempt_number'], 'direct' => $direct, 'outcome' => $operation, - 'observed_at' => now()->toISOString(), + 'observed_at' => now() + ->toISOString(), 'deadline' => $deadline, 'ready' => $directory . '/ready-' . $index, 'continue' => $directory . '/continue', @@ -162,7 +292,11 @@ private function interleavePruning(WorkflowRun $run, array $claim, string $outco usleep(1_000); } foreach ($workers as $worker) { - self::assertSame(0, $worker->getExitCode(), substr($worker->getErrorOutput() . $worker->getOutput(), -8_192)); + self::assertSame( + 0, + $worker->getExitCode(), + substr($worker->getErrorOutput() . $worker->getOutput(), -8_192) + ); } return [ From a9af3ce7241986033235a476b8c7ca0017cadcf7 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 24 Oct 2026 04:35:28 +0000 Subject: [PATCH 4/6] test: bound refusal metadata by the original publication deadline --- .../Unit/V2/ActivityOutcomeRetentionTest.php | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/tests/Unit/V2/ActivityOutcomeRetentionTest.php b/tests/Unit/V2/ActivityOutcomeRetentionTest.php index caa1e8f17..739c9dc8e 100644 --- a/tests/Unit/V2/ActivityOutcomeRetentionTest.php +++ b/tests/Unit/V2/ActivityOutcomeRetentionTest.php @@ -4,6 +4,7 @@ namespace Tests\Unit\V2; +use Illuminate\Database\QueryException; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; use PDO; @@ -224,6 +225,51 @@ public function testSqliteRetentionOwnsTheWriterBeforeReadingItsRunSnapshot(): v self::assertSame(0, $run->historyEvents()->count()); } + public function testSqliteRefusalMetadataKeepsTheOriginalPublicationBudget(): void + { + $connection = DB::connection(); + if ($connection->getDriverName() !== 'sqlite' || $connection->getDatabaseName() === ':memory:') { + $this->markTestSkipped('Publication contention requires two connections to one SQLite file.'); + } + [, $claim, $run] = $this->claimedOriginal(); + $pdo = $connection->getPdo(); + $previousTimeout = (int) $pdo->query('PRAGMA busy_timeout')->fetchColumn(); + $previousJournal = (string) $pdo->query('PRAGMA journal_mode')->fetchColumn(); + self::assertContains(strtolower($previousJournal), ['delete', 'wal', 'truncate', 'persist', 'memory', 'off']); + $pdo->exec('PRAGMA journal_mode=DELETE'); + $pdo->exec('PRAGMA busy_timeout=3000'); + $other = new PDO('sqlite:' . $connection->getDatabaseName()); + $other->exec('BEGIN EXCLUSIVE'); + $started = hrtime(true); + try { + try { + ActivityOutcomeRecorder::record( + $claim['task_id'], + $claim['activity_attempt_id'], + $claim['attempt_number'], + 'must not be published', + null, + 2, + 5, + admissionDeadline: $started + 250_000_000, + ); + self::fail('The held writer cannot admit this outcome.'); + } catch (QueryException $error) { + self::assertInstanceOf(PDOException::class, $error->getPrevious()); + self::assertContains(((int) ($error->getPrevious()->errorInfo[1] ?? 0)) & 255, [5, 6]); + } + $elapsed = (hrtime(true) - $started) / 1_000_000_000; + } finally { + $other->exec('ROLLBACK'); + $pdo->exec('PRAGMA busy_timeout=' . $previousTimeout); + $pdo->exec('PRAGMA journal_mode=' . $previousJournal); + } + + self::assertLessThan(1.0, $elapsed, 'A metadata read must not consume the unrelated three-second SQLite timeout.'); + self::assertSame(RunStatus::Waiting, $run->fresh()->status); + self::assertSame(ActivityAttemptStatus::Running, ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status); + } + private function claimedOriginal(): array { $workflow = WorkflowStub::make(TestRetryWorkflow::class, 'outcome-retention'); From d2088956c2c79f3950ca93fbc87e11004a117a01 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 24 Oct 2026 04:54:27 +0000 Subject: [PATCH 5/6] Keep SQLite refusal reads inside the original outcome admission budget --- src/V2/Support/ActivityOutcomeRecorder.php | 21 ++++++++++++++++--- .../Unit/V2/ActivityOutcomeRetentionTest.php | 17 +++++++++++---- 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/src/V2/Support/ActivityOutcomeRecorder.php b/src/V2/Support/ActivityOutcomeRecorder.php index 22fb56084..f40719d33 100644 --- a/src/V2/Support/ActivityOutcomeRecorder.php +++ b/src/V2/Support/ActivityOutcomeRecorder.php @@ -41,9 +41,13 @@ public static function record( ?string $codec = null, ?int $admissionDeadline = null, ): array { - $refused = self::refusedExpiredAttempt($taskId, $attemptId, $attemptCount, $admissionDeadline); - if ($refused !== null) { - return $refused; + $sqlite = (new ActivityAttempt())->getConnection() + ->getDriverName() === 'sqlite'; + if (! $sqlite) { + $refused = self::refusedExpiredAttempt($taskId, $attemptId, $attemptCount, $admissionDeadline); + if ($refused !== null) { + return $refused; + } } return SqliteTransaction::forTask($taskId, static function () use ( @@ -55,7 +59,18 @@ public static function record( $maxAttempts, $backoffSeconds, $codec, + $sqlite, + $admissionDeadline, ): array { + // Even an unlocked SQLite read can wait behind an exclusive writer. + // Keep the refusal metadata behind the original bounded admission. + if ($sqlite) { + $refused = self::refusedExpiredAttempt($taskId, $attemptId, $attemptCount, $admissionDeadline); + if ($refused !== null) { + return $refused; + } + } + /** @var WorkflowTask|null $task */ $task = WorkflowTask::query() ->lockForUpdate() diff --git a/tests/Unit/V2/ActivityOutcomeRetentionTest.php b/tests/Unit/V2/ActivityOutcomeRetentionTest.php index 739c9dc8e..8c93faacc 100644 --- a/tests/Unit/V2/ActivityOutcomeRetentionTest.php +++ b/tests/Unit/V2/ActivityOutcomeRetentionTest.php @@ -233,8 +233,10 @@ public function testSqliteRefusalMetadataKeepsTheOriginalPublicationBudget(): vo } [, $claim, $run] = $this->claimedOriginal(); $pdo = $connection->getPdo(); - $previousTimeout = (int) $pdo->query('PRAGMA busy_timeout')->fetchColumn(); - $previousJournal = (string) $pdo->query('PRAGMA journal_mode')->fetchColumn(); + $previousTimeout = (int) $pdo->query('PRAGMA busy_timeout') + ->fetchColumn(); + $previousJournal = (string) $pdo->query('PRAGMA journal_mode') + ->fetchColumn(); self::assertContains(strtolower($previousJournal), ['delete', 'wal', 'truncate', 'persist', 'memory', 'off']); $pdo->exec('PRAGMA journal_mode=DELETE'); $pdo->exec('PRAGMA busy_timeout=3000'); @@ -265,9 +267,16 @@ public function testSqliteRefusalMetadataKeepsTheOriginalPublicationBudget(): vo $pdo->exec('PRAGMA journal_mode=' . $previousJournal); } - self::assertLessThan(1.0, $elapsed, 'A metadata read must not consume the unrelated three-second SQLite timeout.'); + self::assertLessThan( + 1.0, + $elapsed, + 'A metadata read must not consume the unrelated three-second SQLite timeout.' + ); self::assertSame(RunStatus::Waiting, $run->fresh()->status); - self::assertSame(ActivityAttemptStatus::Running, ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status); + self::assertSame( + ActivityAttemptStatus::Running, + ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status + ); } private function claimedOriginal(): array From 96a66df8058a63dca422da1639200b6886d65de1 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 24 Oct 2026 05:15:19 +0000 Subject: [PATCH 6/6] Prepare one 2.12.2 patch for failure storage and retention repairs --- CHANGELOG.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a8943dd48..7d22a0c17 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## 2.12.2 - Unreleased +## 2.12.2 - 2026-10-24 - Refuse expired or settled activity callbacks under the original run lock before acquiring task or execution locks. Preserve committed renewals and @@ -10,8 +10,6 @@ within the existing transaction policy and original publication budget. This correction emits no new history event and needs no new event reader. -## 2.12.1 - Unreleased - - Require the upstream TokenGuard string-validation fix on Laravel 11, raising its minimum to 11.56.1. Keep the maintained Laravel 12/13 minimums unchanged. - Record the newly published Composer TokenGuard advisory against the retained