Fence expired activity outcomes against retention

CHANGELOG.md

@@ -1,6 +1,14 @@
 # Changelog
 
-## 2.12.1 - 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
+  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.
 
 - 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.

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"
         },

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.

src/V2/Support/ActivityOutcomeRecorder.php

@@ -41,6 +41,15 @@ public static function record(
         ?string $codec = null,
         ?int $admissionDeadline = null,
     ): array {
+        $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 (
             $taskId,
             $attemptId,
@@ -50,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()
@@ -508,6 +528,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

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);
     }
 
     /**

tests/Support/ActivityOutcomeRetentionProcessTest.php

@@ -0,0 +1,106 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Support;
+
+use Illuminate\Database\Events\QueryExecuted;
+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\ActivityOutcomeRecorder;
+use Workflow\V2\Support\WorkflowRunRetentionCleanup;
+
+final class ActivityOutcomeRetentionProcessTest extends TestCase
+{
+    public function testOutcomeAndRetention(): void
+    {
+        $input = $this->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');
+                }
+            });
+            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']);
+        }
+
+        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);
+    }
+}