From 3477eaccf94e6c087933088069fff72e25a493c4 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 07:44:47 +0000 Subject: [PATCH 01/12] docs: define bounded rejected activity outcome audit --- docs/activity-outcome-rejection-audit.md | 51 ++++++++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 docs/activity-outcome-rejection-audit.md diff --git a/docs/activity-outcome-rejection-audit.md b/docs/activity-outcome-rejection-audit.md new file mode 100644 index 000000000..5344481ed --- /dev/null +++ b/docs/activity-outcome-rejection-audit.md @@ -0,0 +1,51 @@ +# Rejected activity outcome audit + +This is the proposed contract for the remaining history requirement in #878. +It is not a published capability. Existing completion and failure responses +continue to decide whether the business outcome was accepted. + +## Durable fact + +Record `ActivityOutcomeRejected` once for the first refused late outcome of a +recognized original attempt after it loses lease or deadline authority. Bind +the record to its original run, activity, attempt, task, attempt number and +lease owner. Include the attempted outcome (`completed` or `failed`), the +server's refusal reason and the observation timestamp. The first record stays +immutable. Repeated requests do not create additional records. + +Do not store the rejected result, failure message, stack trace, external object +reference, authentication material or an unrestricted request snapshot. +An already committed successful or failed outcome is not a new rejected +outcome merely because its response was retried. + +## Authority and ordering + +- Authenticate and enforce namespace, task, attempt and original owner binding + before any audit write. Unrelated or unknown identities create no record. +- Recheck lost authority under the existing storage transaction and locks. + Select the reason from stored state and original deadlines. A request cannot + nominate its own audit reason or make a valid claim stale. +- Refuse lost authority before resolving or retaining an external payload. + Audit recording never accepts the result, renews a lease, reopens a run, + dispatches work or creates a success receipt. `recorded=false` retains its + existing business meaning. +- Keep prior business history and progress timestamps unchanged. The diagnostic + record advances history sequence/count metadata only. It does not become a + replay step or manufacture business progress. +- Keep retention boundaries. Do not recreate deleted or pruned history, and + serialize the append against pruning. Closed retained runs may acquire this + diagnostic fact without acquiring new execution authority. + +## Required qualification + +Prove original identity, one-record behavior across repeated and concurrent +requests, first-record immutability, transaction rollback, no payload/provider +access, namespace/owner refusals, pruning races and unchanged business state. +Exercise active and closed runs, expired leases, declared retry exhaustion and +authored deadlines on SQLite, MySQL and PostgreSQL. Replay the resulting history +with the published PHP, Python and Rust readers and verify real Waterline +visibility. Existing failed originals remain failed. + +No schema, SDK protocol floor or native/frozen adoption is implied. The event +and payload contract must be coordinated before publication; a source draft or +passing local test does not deliver this requirement. From 11321f75c5a9dfddc5a039a4cce30b35929f5a75 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 07:47:05 +0000 Subject: [PATCH 02/12] test: require bounded history for refused late activity outcomes --- .../V2/ActivityOutcomeRejectionAuditTest.php | 191 ++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php diff --git a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php new file mode 100644 index 000000000..9fa6acf54 --- /dev/null +++ b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php @@ -0,0 +1,191 @@ +freezeTime(); + config(['workflows.v2.task_dispatch_mode' => 'poll']); + } + + public static function lateOutcomes(): array + { + return [ + 'active completion' => [false, 'completed'], + 'closed completion' => [true, 'completed'], + 'active failure' => [false, 'failed'], + 'closed failure' => [true, 'failed'], + ]; + } + + #[DataProvider('lateOutcomes')] + public function testFirstLateOutcomeIsAnImmutableDiagnosticFact(bool $closed, string $outcome): void + { + [$bridge, $claim, $run] = $this->expiredOriginal($closed); + $prefix = $run->historyEvents()->orderBy('sequence')->get()->map->getAttributes()->all(); + $before = $this->businessState($run); + $this->travel(2)->seconds(); + $reply = $this->refusedOutcome($bridge, $claim['activity_attempt_id'], $outcome); + + self::assertFalse($reply['recorded']); + self::assertSame('stale_attempt', $reply['reason']); + self::assertNull($reply['next_task_id']); + $notices = $this->notices($run); + self::assertCount(1, $notices); + $notice = $notices[0]; + self::assertSame($claim['task_id'], $notice->workflow_task_id); + self::assertSame($claim['activity_execution_id'], $notice->payload['activity_execution_id']); + self::assertSame($claim['activity_attempt_id'], $notice->payload['activity_attempt_id']); + self::assertSame($claim['lease_owner'], $notice->payload['lease_owner']); + self::assertSame(1, $notice->payload['attempt_number']); + self::assertSame($outcome, $notice->payload['outcome']); + self::assertSame('stale_attempt', $notice->payload['reason']); + self::assertSame($before, $this->businessState($run)); + self::assertSame($prefix, $run->historyEvents()->orderBy('sequence')->limit(count($prefix)) + ->get()->map->getAttributes()->all()); + foreach (['result', 'failure', 'message', 'stack_trace', 'details', 'task', 'command'] as $field) { + self::assertArrayNotHasKey($field, $notice->payload); + } + + $retained = $notice->getAttributes(); + $this->travel(3)->seconds(); + $this->refusedOutcome($bridge, $claim['activity_attempt_id'], $outcome === 'completed' ? 'failed' : 'completed'); + self::assertCount(1, $this->notices($run)); + self::assertSame($retained, $this->notices($run)[0]->getAttributes()); + self::assertSame($before, $this->businessState($run)); + } + + public function testPrunedRunIsNotRepopulatedByARefusedOutcome(): void + { + [$bridge, $claim, $run] = $this->expiredOriginal(); + $run->forceFill(['details_pruned_at' => now()])->save(); + $before = $this->businessState($run); + $history = $run->historyEvents()->count(); + + self::assertFalse($bridge->complete($claim['activity_attempt_id'], 'late')['recorded']); + self::assertCount(0, $this->notices($run)); + self::assertSame($history, $run->historyEvents()->count()); + self::assertSame($before, $this->businessState($run)); + } + + public function testAuditAppendRollsBackWithItsStorageTransaction(): void + { + [$bridge, $claim, $run] = $this->expiredOriginal(); + $before = $this->businessState($run); + try { + DB::connection($run->getConnectionName())->transaction(function () use ($bridge, $claim, $run): void { + $bridge->complete($claim['activity_attempt_id'], 'late'); + self::assertCount(1, $this->notices($run)); + throw new RuntimeException('rollback diagnostic'); + }); + self::fail('The outer transaction must roll back.'); + } catch (RuntimeException $exception) { + self::assertSame('rollback diagnostic', $exception->getMessage()); + } + self::assertCount(0, $this->notices($run)); + self::assertSame($before, $this->businessState($run)); + } + + public function testRetryingACommittedBusinessOutcomeCreatesNoRejectionAudit(): void + { + [$bridge, $claim, $run] = $this->claimedOriginal(); + self::assertTrue($bridge->complete($claim['activity_attempt_id'], 'accepted')['recorded']); + $history = $run->historyEvents()->count(); + + self::assertFalse($bridge->complete($claim['activity_attempt_id'], 'duplicate')['recorded']); + self::assertCount(0, $this->notices($run)); + self::assertSame($history, $run->historyEvents()->count()); + } + + private function expiredOriginal(bool $closed = false): array + { + [$bridge, $claim, $run] = $this->claimedOriginal(); + $this->travel(301)->seconds(); + self::assertTrue(ActivityTimeoutEnforcer::enforceLeaseExpiry($claim['task_id'])['enforced']); + if ($closed) { + $this->travel(10)->seconds(); + $retry = $bridge->poll(null, null, 1)[0]; + $replacement = $bridge->claimStatus($retry['task_id'], 'replacement-worker'); + self::assertTrue($replacement['claimed']); + self::assertTrue($bridge->complete($replacement['activity_attempt_id'], 'accepted')['recorded']); + $this->runWorkflowTask(); + self::assertTrue($run->fresh()->status->isTerminal()); + } + + return [$bridge, $claim, $run]; + } + + private function claimedOriginal(): array + { + $workflow = WorkflowStub::make(TestRetryWorkflow::class, 'outcome-rejection-audit'); + $workflow->start('Ada'); + $this->runWorkflowTask(); + $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 runWorkflowTask(): void + { + $task = WorkflowTask::query()->where('task_type', TaskType::Workflow) + ->where('status', TaskStatus::Ready)->sole(); + $this->app->call([new RunWorkflowTask($task->id), 'handle']); + } + + private function refusedOutcome(ActivityTaskBridge $bridge, string $attemptId, string $outcome): array + { + return $outcome === 'completed' + ? $bridge->complete($attemptId, new class { + public function __serialize(): array + { + throw new RuntimeException('Rejected output must never be serialized.'); + } + }) + : $bridge->fail($attemptId, ['message' => 'private failure that must not enter audit history']); + } + + private function notices(WorkflowRun $run): array + { + return $run->historyEvents()->where('event_type', 'ActivityOutcomeRejected')->get()->all(); + } + + private function businessState(WorkflowRun $run): array + { + $attributes = $run->fresh()->getAttributes(); + unset($attributes['last_history_sequence'], $attributes['updated_at']); + + return [ + $attributes, + WorkflowTask::query()->where('workflow_run_id', $run->id)->orderBy('id')->get()->map->getAttributes()->all(), + ActivityExecution::query()->where('workflow_run_id', $run->id)->orderBy('id')->get()->map->getAttributes()->all(), + ActivityAttempt::query()->where('workflow_run_id', $run->id)->orderBy('id')->get()->map->getAttributes()->all(), + ]; + } +} From 7d1192f38010d1dc24925414974c632348ce04ec Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 08:21:23 +0000 Subject: [PATCH 03/12] feat: record bounded diagnostics for rejected activity outcomes --- src/V2/Enums/HistoryEventType.php | 1 + src/V2/Models/WorkflowHistoryEvent.php | 17 +-- src/V2/Support/ActivityOutcomeRecorder.php | 18 +++ src/V2/Support/ActivityOutcomeRejection.php | 87 ++++++++++++++ .../Support/HistoryEventPayloadContract.php | 4 + src/V2/Support/HistoryTimeline.php | 42 +++++++ src/V2/Support/RunSummaryProjector.php | 29 +++++ src/V2/Support/RunTimelineProjector.php | 48 +++++--- .../V2/ActivityOutcomeRejectionAuditTest.php | 106 ++++++++++++++---- 9 files changed, 310 insertions(+), 42 deletions(-) create mode 100644 src/V2/Support/ActivityOutcomeRejection.php diff --git a/src/V2/Enums/HistoryEventType.php b/src/V2/Enums/HistoryEventType.php index 8d6cca8d1..aba689e00 100644 --- a/src/V2/Enums/HistoryEventType.php +++ b/src/V2/Enums/HistoryEventType.php @@ -55,6 +55,7 @@ enum HistoryEventType: string case ActivityRetryScheduled = 'ActivityRetryScheduled'; case ActivityCompleted = 'ActivityCompleted'; case ActivityFailed = 'ActivityFailed'; + case ActivityOutcomeRejected = 'ActivityOutcomeRejected'; case ActivityCancelled = 'ActivityCancelled'; case ActivityCancellationAcknowledged = 'ActivityCancellationAcknowledged'; case ActivityTimedOut = 'ActivityTimedOut'; diff --git a/src/V2/Models/WorkflowHistoryEvent.php b/src/V2/Models/WorkflowHistoryEvent.php index 85788867d..96ec5c9b6 100644 --- a/src/V2/Models/WorkflowHistoryEvent.php +++ b/src/V2/Models/WorkflowHistoryEvent.php @@ -131,18 +131,19 @@ public static function record( ]); ExecutionTracing::observeOutcome($lockedRun, $taskModel, $event); - $lockedRun->forceFill([ + $updates = [ 'last_history_sequence' => $sequence, - 'last_progress_at' => $now, - ])->save(); + ]; + if ($eventType !== HistoryEventType::ActivityOutcomeRejected) { + $updates['last_progress_at'] = $now; + } + $lockedRun->forceFill($updates) + ->save(); PendingSignalCloser::closeForTerminalEvent($lockedRun, $event, $taskModel); - $run->forceFill([ - 'last_history_sequence' => $lockedRun->last_history_sequence, - 'last_progress_at' => $lockedRun->last_progress_at, - ]); - $run->syncOriginalAttributes(['last_history_sequence', 'last_progress_at']); + $run->forceFill($updates); + $run->syncOriginalAttributes(array_keys($updates)); LinkedServiceCallCloser::closeForTerminalEvent($lockedRun, $event); ServiceCallWorkflowDelivery::forAdmission($lockedRun, $event); diff --git a/src/V2/Support/ActivityOutcomeRecorder.php b/src/V2/Support/ActivityOutcomeRecorder.php index 275419f4e..b9024b1fa 100644 --- a/src/V2/Support/ActivityOutcomeRecorder.php +++ b/src/V2/Support/ActivityOutcomeRecorder.php @@ -110,6 +110,15 @@ public static function record( || $lockedExecution->current_attempt_id !== $attemptId ) { self::closeAttemptIfStale($run, $attemptId); + ActivityOutcomeRejection::record( + $run, + $lockedExecution, + $task, + $attemptId, + $attemptCount, + $throwable === null ? 'completed' : 'failed', + 'stale_attempt', + ); return self::ignored('stale_attempt'); } @@ -120,6 +129,15 @@ public static function record( $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run); if ($expiredReason !== null) { + ActivityOutcomeRejection::record( + $run, + $lockedExecution, + $task, + $attemptId, + $attemptCount, + $throwable === null ? 'completed' : 'failed', + $expiredReason, + ); return self::ignored($expiredReason); } diff --git a/src/V2/Support/ActivityOutcomeRejection.php b/src/V2/Support/ActivityOutcomeRejection.php new file mode 100644 index 000000000..cbb693bc9 --- /dev/null +++ b/src/V2/Support/ActivityOutcomeRejection.php @@ -0,0 +1,87 @@ +details_pruned_at !== null || ! in_array($outcome, ['completed', 'failed'], true)) { + return null; + } + $attempt = ActivityAttempt::query()->lockForUpdate()->find($attemptId); + if (! $attempt instanceof ActivityAttempt + || $attempt->workflow_task_id !== $task->id + || $attempt->activity_execution_id !== $execution->id + || $attempt->workflow_run_id !== $run->id + || $attempt->attempt_number !== $attemptNumber + || $execution->workflow_run_id !== $run->id + || $task->workflow_run_id !== $run->id + || $task->namespace !== $run->namespace + || ($task->payload['activity_execution_id'] ?? null) !== $execution->id + || ! self::lostAuthority($attempt, $task, $execution, $run, $reason)) { + return null; + } + + $existing = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::ActivityOutcomeRejected) + ->where('workflow_task_id', $task->id) + ->where('payload->activity_attempt_id', $attemptId) + ->first(); + if ($existing instanceof WorkflowHistoryEvent) { + return $existing; + } + + // Passing the stored ID, rather than a task model, deliberately omits + // the generic task snapshot and its unrestricted dispatch diagnostics. + $event = WorkflowHistoryEvent::record($run, HistoryEventType::ActivityOutcomeRejected, [ + 'activity_execution_id' => $execution->id, + 'activity_attempt_id' => $attempt->id, + 'attempt_number' => $attempt->attempt_number, + 'lease_owner' => $attempt->lease_owner, + 'outcome' => $outcome, + 'reason' => $reason, + 'observed_at' => now() + ->toJSON(), + ], (string) $task->id); + RunSummaryProjector::projectActivityOutcomeRejection($run, $event); + RunTimelineProjector::projectActivityOutcomeRejection($run, $event); + + return $event; + } + + private static function lostAuthority( + ActivityAttempt $attempt, + WorkflowTask $task, + ActivityExecution $execution, + WorkflowRun $run, + string $reason, + ): bool { + if ($reason === 'stale_attempt' && $attempt->status === ActivityAttemptStatus::Expired) { + return true; + } + + return $attempt->status === ActivityAttemptStatus::Running + && ActivityAttemptAuthority::expiredReason($task, $execution, now(), $run) === $reason; + } +} diff --git a/src/V2/Support/HistoryEventPayloadContract.php b/src/V2/Support/HistoryEventPayloadContract.php index ccf660a4e..f4e54b8f8 100644 --- a/src/V2/Support/HistoryEventPayloadContract.php +++ b/src/V2/Support/HistoryEventPayloadContract.php @@ -25,6 +25,10 @@ final class HistoryEventPayloadContract * @var array> */ private const PAYLOAD_KEYS = [ + 'ActivityOutcomeRejected' => [ + 'activity_execution_id', 'activity_attempt_id', 'attempt_number', + 'lease_owner', 'outcome', 'reason', 'observed_at', + ], 'CancellationScopeRequested' => [ 'schema', 'workflow_run_id', 'scope_id', 'parent_scope_id', 'request_id', 'cancellation', ], diff --git a/src/V2/Support/HistoryTimeline.php b/src/V2/Support/HistoryTimeline.php index 944287aab..c8a0710ea 100644 --- a/src/V2/Support/HistoryTimeline.php +++ b/src/V2/Support/HistoryTimeline.php @@ -138,6 +138,45 @@ public static function fromWorkerTerminalEvent(WorkflowHistoryEvent $event): arr return self::mapEvent($event, collect(), collect(), collect(), collect(), $failures); } + /** + * @return array + */ + public static function fromActivityOutcomeRejection(WorkflowHistoryEvent $event): array + { + if ($event->event_type !== HistoryEventType::ActivityOutcomeRejected) { + throw new \LogicException('Rejection timeline requires an ActivityOutcomeRejected event.'); + } + $payload = is_array($event->payload) ? $event->payload : []; + + return [ + 'id' => $event->id, + 'sequence' => $event->sequence, + 'type' => $event->event_type->value, + 'kind' => 'activity', + 'entry_kind' => 'point', + 'source_kind' => 'activity_execution', + 'source_id' => $payload['activity_execution_id'] ?? null, + 'summary' => sprintf( + 'Rejected %s outcome from activity attempt %s: %s.', + $payload['outcome'] ?? 'unknown', + $payload['attempt_number'] ?? 'unknown', + $payload['reason'] ?? 'unknown', + ), + 'recorded_at' => self::timestamp($event->recorded_at), + 'command_id' => null, + 'command_sequence' => null, + 'task_id' => $event->workflow_task_id, + 'activity_execution_id' => $payload['activity_execution_id'] ?? null, + 'timer_id' => null, + 'failure_id' => null, + 'activity' => [ + 'id' => $payload['activity_execution_id'] ?? null, + 'attempt_id' => $payload['activity_attempt_id'] ?? null, + 'attempt_count' => $payload['attempt_number'] ?? null, + ], + ]; + } + private static function mapEvent( WorkflowHistoryEvent $event, Collection $commands, @@ -146,6 +185,9 @@ private static function mapEvent( Collection $timers, Collection $failures, ): array { + if ($event->event_type === HistoryEventType::ActivityOutcomeRejected) { + return self::fromActivityOutcomeRejection($event); + } /** @var array $payload */ $payload = is_array($event->payload) ? $event->payload : []; $commandId = self::stringValue($event->workflow_command_id) diff --git a/src/V2/Support/RunSummaryProjector.php b/src/V2/Support/RunSummaryProjector.php index f99d3db7f..dfbe9eb26 100644 --- a/src/V2/Support/RunSummaryProjector.php +++ b/src/V2/Support/RunSummaryProjector.php @@ -96,6 +96,35 @@ public static function projectBufferedSignal( return $summary; } + public static function projectActivityOutcomeRejection(WorkflowRun $run, WorkflowHistoryEvent $event): void + { + if ($event->workflow_run_id !== $run->id + || $event->event_type !== HistoryEventType::ActivityOutcomeRejected) { + throw new \LogicException('Rejection counters must belong to the original workflow run.'); + } + $summary = self::summaryModel()::query()->find($run->id); + if (! $summary instanceof WorkflowRunSummary + || $summary->projection_schema_version !== self::SCHEMA_VERSION + || (int) $summary->history_event_count !== (int) $event->sequence - 1 + || (int) $summary->history_size_bytes <= 0 + || (int) $summary->history_fan_out < 0) { + return; + } + $budget = HistoryBudget::fromCounters( + (int) $event->sequence, + (int) $summary->history_size_bytes + HistoryBudget::sizeOfEvent($event), + (int) $summary->history_fan_out, + ); + // Preserve the business timestamp and every routing/wait field. A + // diagnostic append must not look like workflow progress or repair. + self::summaryModel()::query()->whereKey($run->id)->toBase()->update([ + 'history_event_count' => $budget['history_event_count'], + 'history_size_bytes' => $budget['history_size_bytes'], + 'continue_as_new_recommended' => $budget['continue_as_new_recommended'], + 'history_budget_pressure' => $budget['pressure'], + ]); + } + public static function project(WorkflowRun $run): WorkflowRunSummary { $claimedTask = WorkflowTaskClaimProjectionContext::taskFor($run); diff --git a/src/V2/Support/RunTimelineProjector.php b/src/V2/Support/RunTimelineProjector.php index 6738bc323..bdaf58819 100644 --- a/src/V2/Support/RunTimelineProjector.php +++ b/src/V2/Support/RunTimelineProjector.php @@ -143,22 +143,14 @@ public static function projectWorkerTerminalEvent( WorkflowRun $run, WorkflowHistoryEvent $event, ): WorkflowTimelineEntry { - if ($event->workflow_run_id !== $run->id) { - throw new \LogicException('Timeline event must belong to the projected workflow run.'); - } - - $historyEventId = (string) $event->id; - $row = self::upsertEntry( - $run, - self::entryModel(), - self::projectionId($run->id, $historyEventId), - $historyEventId, - HistoryTimeline::fromWorkerTerminalEvent($event), - ); - - $run->unsetRelation('timelineEntries'); + return self::projectHistoryPoint($run, $event, HistoryTimeline::fromWorkerTerminalEvent($event)); + } - return $row; + public static function projectActivityOutcomeRejection( + WorkflowRun $run, + WorkflowHistoryEvent $event, + ): WorkflowTimelineEntry { + return self::projectHistoryPoint($run, $event, HistoryTimeline::fromActivityOutcomeRejection($event)); } /** @@ -230,6 +222,32 @@ public static function driftStatusForRun(WorkflowRun $run): array ]; } + /** + * @param array $entry + */ + private static function projectHistoryPoint( + WorkflowRun $run, + WorkflowHistoryEvent $event, + array $entry, + ): WorkflowTimelineEntry { + if ($event->workflow_run_id !== $run->id) { + throw new \LogicException('Timeline event must belong to the projected workflow run.'); + } + + $historyEventId = (string) $event->id; + $row = self::upsertEntry( + $run, + self::entryModel(), + self::projectionId($run->id, $historyEventId), + $historyEventId, + $entry, + ); + + $run->unsetRelation('timelineEntries'); + + return $row; + } + /** * @param class-string $entryModel * @param array $entry diff --git a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php index 9fa6acf54..b8a657e01 100644 --- a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php +++ b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php @@ -16,10 +16,13 @@ use Workflow\V2\Jobs\RunWorkflowTask; use Workflow\V2\Models\ActivityAttempt; use Workflow\V2\Models\ActivityExecution; -use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; +use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\Models\WorkflowTimelineEntry; use Workflow\V2\Support\ActivityTimeoutEnforcer; +use Workflow\V2\Support\HistoryBudget; +use Workflow\V2\Support\RunTimelineProjector; use Workflow\V2\WorkflowStub; final class ActivityOutcomeRejectionAuditTest extends TestCase @@ -29,7 +32,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 lateOutcomes(): array @@ -46,9 +51,14 @@ public static function lateOutcomes(): array public function testFirstLateOutcomeIsAnImmutableDiagnosticFact(bool $closed, string $outcome): void { [$bridge, $claim, $run] = $this->expiredOriginal($closed); - $prefix = $run->historyEvents()->orderBy('sequence')->get()->map->getAttributes()->all(); + $prefix = $run->historyEvents() + ->orderBy('sequence') + ->get() + ->map->getAttributes() + ->all(); $before = $this->businessState($run); - $this->travel(2)->seconds(); + $this->travel(2) + ->seconds(); $reply = $this->refusedOutcome($bridge, $claim['activity_attempt_id'], $outcome); self::assertFalse($reply['recorded']); @@ -66,14 +76,21 @@ public function testFirstLateOutcomeIsAnImmutableDiagnosticFact(bool $closed, st self::assertSame('stale_attempt', $notice->payload['reason']); self::assertSame($before, $this->businessState($run)); self::assertSame($prefix, $run->historyEvents()->orderBy('sequence')->limit(count($prefix)) - ->get()->map->getAttributes()->all()); + ->get() + ->map->getAttributes() + ->all()); foreach (['result', 'failure', 'message', 'stack_trace', 'details', 'task', 'command'] as $field) { self::assertArrayNotHasKey($field, $notice->payload); } $retained = $notice->getAttributes(); - $this->travel(3)->seconds(); - $this->refusedOutcome($bridge, $claim['activity_attempt_id'], $outcome === 'completed' ? 'failed' : 'completed'); + $this->travel(3) + ->seconds(); + $this->refusedOutcome( + $bridge, + $claim['activity_attempt_id'], + $outcome === 'completed' ? 'failed' : 'completed' + ); self::assertCount(1, $this->notices($run)); self::assertSame($retained, $this->notices($run)[0]->getAttributes()); self::assertSame($before, $this->businessState($run)); @@ -82,9 +99,12 @@ public function testFirstLateOutcomeIsAnImmutableDiagnosticFact(bool $closed, st public function testPrunedRunIsNotRepopulatedByARefusedOutcome(): void { [$bridge, $claim, $run] = $this->expiredOriginal(); - $run->forceFill(['details_pruned_at' => now()])->save(); + $run->forceFill([ + 'details_pruned_at' => now(), + ])->save(); $before = $this->businessState($run); - $history = $run->historyEvents()->count(); + $history = $run->historyEvents() + ->count(); self::assertFalse($bridge->complete($claim['activity_attempt_id'], 'late')['recorded']); self::assertCount(0, $this->notices($run)); @@ -114,20 +134,56 @@ public function testRetryingACommittedBusinessOutcomeCreatesNoRejectionAudit(): { [$bridge, $claim, $run] = $this->claimedOriginal(); self::assertTrue($bridge->complete($claim['activity_attempt_id'], 'accepted')['recorded']); - $history = $run->historyEvents()->count(); + $history = $run->historyEvents() + ->count(); self::assertFalse($bridge->complete($claim['activity_attempt_id'], 'duplicate')['recorded']); self::assertCount(0, $this->notices($run)); self::assertSame($history, $run->historyEvents()->count()); } + public function testDiagnosticAppendKeepsBoundedCountersAndTimelineCurrent(): void + { + [$bridge, $claim, $run] = $this->expiredOriginal(closed: true); + self::assertFalse( + RunTimelineProjector::driftStatusForRun($run->fresh())['stale'], + 'The original timeline must already be current.' + ); + $summary = WorkflowRunSummary::query()->findOrFail($run->id); + $summaryBefore = $summary->getAttributes(); + $prefix = WorkflowTimelineEntry::query()->where('workflow_run_id', $run->id)->orderBy('sequence') + ->get() + ->map->getAttributes() + ->all(); + $this->travel(2) + ->seconds(); + $bridge->complete($claim['activity_attempt_id'], 'late'); + + $summary = $summary->fresh(); + self::assertTrue(HistoryBudget::hasCompleteSummary($run->fresh(), $summary)); + self::assertSame((int) $summaryBefore['history_event_count'] + 1, $summary->history_event_count); + self::assertGreaterThan((int) $summaryBefore['history_size_bytes'], $summary->history_size_bytes); + self::assertEquals($summaryBefore['updated_at'], $summary->getRawOriginal('updated_at')); + foreach (['status', 'next_task_id', 'last_failure_id', 'open_wait_count'] as $field) { + self::assertEquals($summaryBefore[$field] ?? null, $summary->getRawOriginal($field)); + } + $rows = WorkflowTimelineEntry::query()->where('workflow_run_id', $run->id)->orderBy('sequence')->get(); + self::assertCount(count($prefix) + 1, $rows); + self::assertSame($prefix, $rows->take(count($prefix))->map->getAttributes()->all()); + self::assertSame('ActivityOutcomeRejected', $rows->last()->type); + self::assertStringContainsString('stale_attempt', $rows->last()->summary); + self::assertFalse(RunTimelineProjector::driftStatusForRun($run->fresh())['stale']); + } + private function expiredOriginal(bool $closed = false): array { [$bridge, $claim, $run] = $this->claimedOriginal(); - $this->travel(301)->seconds(); + $this->travel(301) + ->seconds(); self::assertTrue(ActivityTimeoutEnforcer::enforceLeaseExpiry($claim['task_id'])['enforced']); if ($closed) { - $this->travel(10)->seconds(); + $this->travel(10) + ->seconds(); $retry = $bridge->poll(null, null, 1)[0]; $replacement = $bridge->claimStatus($retry['task_id'], 'replacement-worker'); self::assertTrue($replacement['claimed']); @@ -162,30 +218,42 @@ private function runWorkflowTask(): void private function refusedOutcome(ActivityTaskBridge $bridge, string $attemptId, string $outcome): array { return $outcome === 'completed' - ? $bridge->complete($attemptId, new class { + ? $bridge->complete($attemptId, new class() { public function __serialize(): array { throw new RuntimeException('Rejected output must never be serialized.'); } }) - : $bridge->fail($attemptId, ['message' => 'private failure that must not enter audit history']); + : $bridge->fail($attemptId, [ + 'message' => 'private failure that must not enter audit history', + ]); } private function notices(WorkflowRun $run): array { - return $run->historyEvents()->where('event_type', 'ActivityOutcomeRejected')->get()->all(); + return $run->historyEvents() + ->where('event_type', 'ActivityOutcomeRejected') + ->get() + ->all(); } private function businessState(WorkflowRun $run): array { - $attributes = $run->fresh()->getAttributes(); + $attributes = $run->fresh() + ->getAttributes(); unset($attributes['last_history_sequence'], $attributes['updated_at']); return [ $attributes, - WorkflowTask::query()->where('workflow_run_id', $run->id)->orderBy('id')->get()->map->getAttributes()->all(), - ActivityExecution::query()->where('workflow_run_id', $run->id)->orderBy('id')->get()->map->getAttributes()->all(), - ActivityAttempt::query()->where('workflow_run_id', $run->id)->orderBy('id')->get()->map->getAttributes()->all(), + WorkflowTask::query()->where('workflow_run_id', $run->id)->orderBy('id')->get() + ->map->getAttributes() + ->all(), + ActivityExecution::query()->where('workflow_run_id', $run->id)->orderBy('id')->get() + ->map->getAttributes() + ->all(), + ActivityAttempt::query()->where('workflow_run_id', $run->id)->orderBy('id')->get() + ->map->getAttributes() + ->all(), ]; } } From 58d78bd8a8bc54462e5583135cd836f3d54f7f9e Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 08:31:49 +0000 Subject: [PATCH 04/12] docs: link the rejected outcome audit requirement --- docs/activity-outcome-rejection-audit.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/activity-outcome-rejection-audit.md b/docs/activity-outcome-rejection-audit.md index 5344481ed..fc809c998 100644 --- a/docs/activity-outcome-rejection-audit.md +++ b/docs/activity-outcome-rejection-audit.md @@ -1,6 +1,7 @@ # Rejected activity outcome audit -This is the proposed contract for the remaining history requirement in #878. +This is the proposed contract for the remaining history requirement in +[Workflow issue 878](https://github.com/durable-workflow/workflow/issues/878). It is not a published capability. Existing completion and failure responses continue to decide whether the business outcome was accepted. From dc6048c4fa398f46f1b15f827ca9954c682fb313 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 08:56:17 +0000 Subject: [PATCH 05/12] test: retain rejected activity outcome cold replay evidence --- ...activity-rejected-outcome-cold-replay.json | 100 ++++++++++++++++++ 1 file changed, 100 insertions(+) create mode 100644 tests/Fixtures/V2/ReplayRegression/activity-rejected-outcome-cold-replay.json diff --git a/tests/Fixtures/V2/ReplayRegression/activity-rejected-outcome-cold-replay.json b/tests/Fixtures/V2/ReplayRegression/activity-rejected-outcome-cold-replay.json new file mode 100644 index 000000000..8eda29080 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/activity-rejected-outcome-cold-replay.json @@ -0,0 +1,100 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "activity-rejected-outcome-cold-replay", + "protocol_version": "1.0", + "bindings": ["php"], + "consumers": ["workflow-fiber-runner"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestHandledFailureWorkflow", + "arguments": [], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-10-22T01:00:00.000000Z" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "recorded_at": "2026-10-22T01:00:00.000001Z", + "payload": { + "sequence": 1, + "activity_execution_id": "01j00000000000000000000001", + "activity_class": "Tests\\Fixtures\\V2\\TestFailingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestFailingActivity" + } + }, + { + "sequence": 3, + "event_type": "ActivityStarted", + "recorded_at": "2026-10-22T01:00:00.000002Z", + "payload": { + "sequence": 1, + "activity_execution_id": "01j00000000000000000000001", + "activity_attempt_id": "01j0000000000000000000000a", + "activity_class": "Tests\\Fixtures\\V2\\TestFailingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestFailingActivity", + "attempt_number": 1 + } + }, + { + "sequence": 4, + "event_type": "ActivityFailed", + "recorded_at": "2026-10-22T01:00:30.000002Z", + "payload": { + "sequence": 1, + "activity_execution_id": "01j00000000000000000000001", + "activity_attempt_id": "01j0000000000000000000000a", + "activity_class": "Tests\\Fixtures\\V2\\TestFailingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestFailingActivity", + "attempt_number": 1, + "failure_id": "01j0000000000000000000000f", + "failure_category": "internal", + "non_retryable": false, + "failure_reason": "lease_expired", + "lease_expires_at": "2026-10-22T01:00:30.000002Z", + "lease_owner": "original-worker", + "exception_class": "Workflow\\V2\\Exceptions\\ActivityLeaseExpiredException", + "message": "Activity Tests\\Fixtures\\V2\\TestFailingActivity lost its original task lease before the attempt settled." + } + }, + { + "sequence": 5, + "event_type": "ActivityOutcomeRejected", + "recorded_at": "2026-10-22T01:00:30.500000Z", + "payload": { + "activity_execution_id": "01j00000000000000000000001", + "activity_attempt_id": "01j0000000000000000000000a", + "attempt_number": 1, + "lease_owner": "original-worker", + "outcome": "completed", + "reason": "stale_attempt", + "observed_at": "2026-10-22T01:00:30.500000Z" + } + }, + { + "sequence": 6, + "event_type": "ActivityCompleted", + "recorded_at": "2026-10-22T01:00:31.000000Z", + "payload": { + "sequence": 2, + "activity_execution_id": "01j00000000000000000000002", + "activity_attempt_id": "01j0000000000000000000000b", + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "attempt_number": 1, + "result": "wwHioz3/VYAiNwoiSGVsbG8sIFJlY292ZXJlZCE=", + "payload_codec": "avro" + } + } + ], + "expected": { + "completed": true, + "result": "Hello, Recovered!", + "commands": [{"type": "complete_workflow"}] + } +} From 7e1703fb383d3eb9542e73ae5b77d227ae87e099 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 09:04:26 +0000 Subject: [PATCH 06/12] test: expose missing rejection notices after result preparation --- .../V2/ActivityOutcomeRejectionAuditTest.php | 64 +++++++++++++++++++ 1 file changed, 64 insertions(+) diff --git a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php index b8a657e01..b4a47b1e2 100644 --- a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php +++ b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php @@ -4,13 +4,17 @@ namespace Tests\Unit\V2; +use Carbon\CarbonInterface; use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Date; use Illuminate\Support\Facades\Queue; use PHPUnit\Framework\Attributes\DataProvider; use RuntimeException; use Tests\Fixtures\V2\TestRetryWorkflow; use Tests\TestCase; use Workflow\V2\Contracts\ActivityTaskBridge; +use Workflow\V2\Contracts\ExternalPayloadStorageDriver; +use Workflow\V2\Contracts\ExternalPayloadStoragePolicy; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; use Workflow\V2\Jobs\RunWorkflowTask; @@ -175,6 +179,66 @@ public function testDiagnosticAppendKeepsBoundedCountersAndTimelineCurrent(): vo self::assertFalse(RunTimelineProjector::driftStatusForRun($run->fresh())['stale']); } + public static function publicationDeadlines(): array + { + return [ + 'lease boundary' => ['lease_expired', false], + 'activity boundary' => ['start_to_close_timeout', false], + 'run boundary' => ['run_timeout', false], + 'completed request converted by payload guard' => ['lease_expired', true], + ]; + } + + #[DataProvider('publicationDeadlines')] + public function testRefusalDuringPreparationRetainsOriginalOutcome(string $reason, bool $payloadLimit): void + { + [$bridge, $claim, $run] = $this->claimedOriginal(); + $task = WorkflowTask::query()->findOrFail($claim['task_id']); + $boundary = now()->addSecond(); + if ($reason === 'lease_expired') { + $boundary = $task->lease_expires_at; + } elseif ($reason === 'start_to_close_timeout') { + ActivityExecution::query()->findOrFail($claim['activity_execution_id'])->forceFill([ + 'close_deadline_at' => $boundary, + ])->save(); + } else { + $run->forceFill(['run_deadline_at' => $boundary])->save(); + } + if ($payloadLimit) { + config(['workflows.v2.structural_limits.payload_size_bytes' => 1]); + } + $this->app->instance(ExternalPayloadStoragePolicy::class, new class($boundary) implements ExternalPayloadStoragePolicy { + public function __construct(private readonly CarbonInterface $boundary) + { + } + + public function driverFor(?string $namespace): ?ExternalPayloadStorageDriver + { + Date::setTestNow($this->boundary); + + return null; + } + + public function thresholdBytesFor(?string $namespace): ?int + { + return null; + } + }); + $before = $this->businessState($run); + + $reply = $bridge->complete($claim['activity_attempt_id'], 'discarded-private-result'); + + self::assertFalse($reply['recorded']); + self::assertSame($reason, $reply['reason']); + self::assertCount(1, $this->notices($run)); + $notice = $this->notices($run)[0]; + self::assertSame('completed', $notice->payload['outcome']); + self::assertSame($reason, $notice->payload['reason']); + self::assertSame($claim['activity_attempt_id'], $notice->payload['activity_attempt_id']); + self::assertStringNotContainsString('discarded-private-result', json_encode($notice->payload)); + self::assertSame($before, $this->businessState($run)); + } + private function expiredOriginal(bool $closed = false): array { [$bridge, $claim, $run] = $this->claimedOriginal(); From 5def215c3c1cd2a94e91c7cfea99506c57b895b4 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 09:21:40 +0000 Subject: [PATCH 07/12] feat: audit expired activity outcomes at publication boundaries --- src/V2/Support/ActivityOutcomeRecorder.php | 80 ++++++++++++++----- tests/Feature/V2/V2ActivityTaskBridgeTest.php | 27 ++++++- .../V2/ActivityOutcomeRejectionAuditTest.php | 22 +++-- 3 files changed, 99 insertions(+), 30 deletions(-) diff --git a/src/V2/Support/ActivityOutcomeRecorder.php b/src/V2/Support/ActivityOutcomeRecorder.php index b9024b1fa..d426c6a9c 100644 --- a/src/V2/Support/ActivityOutcomeRecorder.php +++ b/src/V2/Support/ActivityOutcomeRecorder.php @@ -76,6 +76,7 @@ public static function record( ->lockForUpdate() ->findOrFail($lockedExecution->workflow_run_id); + $attemptedOutcome = $throwable === null ? 'completed' : 'failed'; $abandoned = ActivityAbandonment::allows($run, $lockedExecution); if (in_array($run->status, [RunStatus::Cancelled, RunStatus::Terminated], true) && ! $abandoned) { $reason = $run->status === RunStatus::Terminated @@ -116,7 +117,7 @@ public static function record( $task, $attemptId, $attemptCount, - $throwable === null ? 'completed' : 'failed', + $attemptedOutcome, 'stale_attempt', ); @@ -127,18 +128,16 @@ public static function record( return self::ignored('activity_deadline_elapsed'); } - $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run); - if ($expiredReason !== null) { - ActivityOutcomeRejection::record( - $run, - $lockedExecution, - $task, - $attemptId, - $attemptCount, - $throwable === null ? 'completed' : 'failed', - $expiredReason, - ); - return self::ignored($expiredReason); + $refusal = self::expiredOutcome( + $run, + $lockedExecution, + $task, + $attemptId, + $attemptCount, + $attemptedOutcome + ); + if ($refusal !== null) { + return $refusal; } $runCodec = is_string($run->payload_codec) && $run->payload_codec !== '' @@ -186,9 +185,16 @@ public static function record( } // Payload preparation can outlive a claim even while its rows stay locked. - $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run); - if ($expiredReason !== null) { - return self::ignored($expiredReason); + $refusal = self::expiredOutcome( + $run, + $lockedExecution, + $task, + $attemptId, + $attemptCount, + $attemptedOutcome + ); + if ($refusal !== null) { + return $refusal; } if (in_array($run->status, [RunStatus::Completed, RunStatus::Failed], true)) { @@ -228,9 +234,16 @@ public static function record( $parallelMetadata = ParallelChildGroup::payloadForPath($parallelMetadataPath); $resolutionEvent = null; - $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run); - if ($expiredReason !== null) { - return self::ignored($expiredReason); + $refusal = self::expiredOutcome( + $run, + $lockedExecution, + $task, + $attemptId, + $attemptCount, + $attemptedOutcome + ); + if ($refusal !== null) { + return $refusal; } if ($throwable === null) { @@ -633,6 +646,35 @@ private static function ignored(string $reason): array ]; } + /** + * @return array{recorded: bool, reason: string|null, next_task: WorkflowTask|null}|null + */ + private static function expiredOutcome( + WorkflowRun $run, + ActivityExecution $execution, + WorkflowTask $task, + string $attemptId, + int $attemptCount, + string $attemptedOutcome, + ): ?array { + $reason = ActivityAttemptAuthority::expiredReason($task, $execution, now(), $run); + if ($reason === null) { + return null; + } + + ActivityOutcomeRejection::record( + $run, + $execution, + $task, + $attemptId, + $attemptCount, + $attemptedOutcome, + $reason + ); + + return self::ignored($reason); + } + /** * @return array */ diff --git a/tests/Feature/V2/V2ActivityTaskBridgeTest.php b/tests/Feature/V2/V2ActivityTaskBridgeTest.php index e7a8f9369..6fa85a67d 100644 --- a/tests/Feature/V2/V2ActivityTaskBridgeTest.php +++ b/tests/Feature/V2/V2ActivityTaskBridgeTest.php @@ -341,7 +341,7 @@ public function testExpiredCurrentLeaseCannotCompleteBeforeRepair(): void $this->assertFalse($outcome['recorded']); $this->assertSame('lease_expired', $outcome['reason']); - $this->assertSame($before, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + $this->assertRejectedOutcomeDiagnostic($run, $claim, $before, 'completed', 'lease_expired'); $this->assertSame(TaskStatus::Leased, $task->fresh()->status); $this->assertSame(ActivityStatus::Running, $execution->fresh()->status); $this->assertSame( @@ -366,7 +366,7 @@ public function testExpiredCurrentLeaseCannotRecordApplicationFailureBeforeRepai $this->assertFalse($outcome['recorded']); $this->assertSame('lease_expired', $outcome['reason']); - $this->assertSame($before, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + $this->assertRejectedOutcomeDiagnostic($run, $claim, $before, 'failed', 'lease_expired'); $this->assertSame(ActivityStatus::Running, $execution->fresh()->status); $this->assertSame(TaskStatus::Leased, $task->fresh()->status); } finally { @@ -449,7 +449,7 @@ public function thresholdBytesFor(?string $namespace): ?int $this->assertFalse($outcome['recorded']); $this->assertSame('lease_expired', $outcome['reason']); - $this->assertSame($before, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + $this->assertRejectedOutcomeDiagnostic($run, $claim, $before, 'completed', 'lease_expired'); $this->assertSame(TaskStatus::Leased, $task->fresh()->status); $this->assertSame(ActivityStatus::Running, $execution->fresh()->status); } finally { @@ -501,7 +501,7 @@ public function testExpiredExecutionDeadlineFencesOutcomeAndRenewalWithLiveLease $this->assertFalse($fail['recorded']); $this->assertSame($reason, $fail['reason']); $this->assertTrue($task->fresh()->lease_expires_at->equalTo($expiry)); - $this->assertSame($before, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + $this->assertRejectedOutcomeDiagnostic($run, $claim, $before, 'completed', $reason); $timeout = ActivityTimeoutEnforcer::enforce($execution->id); $this->assertTrue($timeout['enforced']); @@ -1627,6 +1627,25 @@ public function thresholdBytesFor(?string $namespace): ?int ); } + private function assertRejectedOutcomeDiagnostic( + WorkflowRun $run, + array $claim, + int $prefixSize, + string $outcome, + string $reason + ): void { + $this->assertSame($prefixSize + 1, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + $event = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::ActivityOutcomeRejected)->sole(); + $this->assertSame($claim['task_id'], $event->workflow_task_id); + $this->assertSame($claim['activity_attempt_id'], $event->payload['activity_attempt_id']); + $this->assertSame($outcome, $event->payload['outcome']); + $this->assertSame($reason, $event->payload['reason']); + foreach (['result', 'exception', 'message', 'stack_trace', 'task', 'command'] as $field) { + $this->assertArrayNotHasKey($field, $event->payload); + } + } + private function assertExternalArgumentsEnvelope(mixed $arguments): void { $this->assertIsArray($arguments); diff --git a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php index b4a47b1e2..c0b20f53f 100644 --- a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php +++ b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php @@ -5,8 +5,8 @@ namespace Tests\Unit\V2; use Carbon\CarbonInterface; -use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Date; +use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; use PHPUnit\Framework\Attributes\DataProvider; use RuntimeException; @@ -194,7 +194,8 @@ public function testRefusalDuringPreparationRetainsOriginalOutcome(string $reaso { [$bridge, $claim, $run] = $this->claimedOriginal(); $task = WorkflowTask::query()->findOrFail($claim['task_id']); - $boundary = now()->addSecond(); + $boundary = now() + ->addSecond(); if ($reason === 'lease_expired') { $boundary = $task->lease_expires_at; } elseif ($reason === 'start_to_close_timeout') { @@ -202,14 +203,21 @@ public function testRefusalDuringPreparationRetainsOriginalOutcome(string $reaso 'close_deadline_at' => $boundary, ])->save(); } else { - $run->forceFill(['run_deadline_at' => $boundary])->save(); + $run->forceFill([ + 'run_deadline_at' => $boundary, + ])->save(); } if ($payloadLimit) { - config(['workflows.v2.structural_limits.payload_size_bytes' => 1]); + config([ + 'workflows.v2.structural_limits.payload_size_bytes' => 1, + ]); } - $this->app->instance(ExternalPayloadStoragePolicy::class, new class($boundary) implements ExternalPayloadStoragePolicy { - public function __construct(private readonly CarbonInterface $boundary) - { + $this->app->instance(ExternalPayloadStoragePolicy::class, new class( + $boundary + ) implements ExternalPayloadStoragePolicy { + public function __construct( + private readonly CarbonInterface $boundary + ) { } public function driverFor(?string $namespace): ?ExternalPayloadStorageDriver From 18419ee52759d4cadbfa5723c9c4ecf9b3f4e5f2 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 10:18:20 +0000 Subject: [PATCH 08/12] test: require exact ownership for rejected activity audit --- .../V2/ActivityOutcomeRejectionAuditTest.php | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php index c0b20f53f..a323f59f6 100644 --- a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php +++ b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php @@ -20,13 +20,16 @@ use Workflow\V2\Jobs\RunWorkflowTask; use Workflow\V2\Models\ActivityAttempt; use Workflow\V2\Models\ActivityExecution; +use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Models\WorkflowTimelineEntry; use Workflow\V2\Support\ActivityTimeoutEnforcer; +use Workflow\V2\Support\ActivityOutcomeRejection; use Workflow\V2\Support\HistoryBudget; use Workflow\V2\Support\RunTimelineProjector; +use Workflow\V2\Support\SqliteTransaction; use Workflow\V2\WorkflowStub; final class ActivityOutcomeRejectionAuditTest extends TestCase @@ -116,6 +119,55 @@ public function testPrunedRunIsNotRepopulatedByARefusedOutcome(): void self::assertSame($before, $this->businessState($run)); } + public function testDiagnosticRequiresExactAttemptIdDespiteCaseInsensitiveStorage(): void + { + [, $claim, $run] = $this->expiredOriginal(); + $alias = strtolower($claim['activity_attempt_id']); + if ($alias === $claim['activity_attempt_id']) { + $alias = strtoupper($claim['activity_attempt_id']); + } + self::assertNotSame($claim['activity_attempt_id'], $alias); + $lookup = ActivityAttempt::query()->find($alias); + if (DB::connection()->getDriverName() === 'mysql') { + self::assertInstanceOf(ActivityAttempt::class, $lookup); + self::assertSame($claim['activity_attempt_id'], $lookup->id); + } + $before = $this->businessState($run); + + self::assertNull($this->recordDiagnostic($claim, $run, $alias)); + self::assertCount(0, $this->notices($run)); + self::assertSame($before, $this->businessState($run)); + + $canonical = $this->recordDiagnostic($claim, $run, $claim['activity_attempt_id']); + self::assertInstanceOf(WorkflowHistoryEvent::class, $canonical); + self::assertSame($claim['activity_attempt_id'], $canonical->payload['activity_attempt_id']); + self::assertSame($claim['lease_owner'], $canonical->payload['lease_owner']); + self::assertCount(1, $this->notices($run)); + self::assertSame($before, $this->businessState($run)); + } + + public static function missingOriginalOwners(): array + { + return [ + 'unknown original owner' => [null], + 'empty original owner' => [''], + ]; + } + + #[DataProvider('missingOriginalOwners')] + public function testDiagnosticDoesNotInventMissingOriginalOwnership(?string $owner): void + { + [, $claim, $run] = $this->expiredOriginal(); + ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->forceFill([ + 'lease_owner' => $owner, + ])->save(); + $before = $this->businessState($run); + + self::assertNull($this->recordDiagnostic($claim, $run, $claim['activity_attempt_id'])); + self::assertCount(0, $this->notices($run)); + self::assertSame($before, $this->businessState($run)); + } + public function testAuditAppendRollsBackWithItsStorageTransaction(): void { [$bridge, $claim, $run] = $this->expiredOriginal(); @@ -301,6 +353,25 @@ public function __serialize(): array ]); } + private function recordDiagnostic(array $claim, WorkflowRun $run, string $attemptId): ?WorkflowHistoryEvent + { + return SqliteTransaction::forTask($claim['task_id'], static function () use ($claim, $run, $attemptId) { + $task = WorkflowTask::query()->lockForUpdate()->findOrFail($claim['task_id']); + $execution = ActivityExecution::query()->lockForUpdate()->findOrFail($claim['activity_execution_id']); + $lockedRun = WorkflowRun::query()->lockForUpdate()->findOrFail($run->id); + + return ActivityOutcomeRejection::record( + $lockedRun, + $execution, + $task, + $attemptId, + $claim['attempt_number'], + 'completed', + 'stale_attempt', + ); + }); + } + private function notices(WorkflowRun $run): array { return $run->historyEvents() From 86ea5d675c430024007b2522bd825edcb2872241 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 10:28:14 +0000 Subject: [PATCH 09/12] fix: require original identity for rejected outcome diagnostics --- docs/activity-outcome-rejection-audit.md | 4 ++++ src/V2/Support/ActivityOutcomeRejection.php | 3 +++ tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php | 2 +- 3 files changed, 8 insertions(+), 1 deletion(-) diff --git a/docs/activity-outcome-rejection-audit.md b/docs/activity-outcome-rejection-audit.md index fc809c998..a1325a6c8 100644 --- a/docs/activity-outcome-rejection-audit.md +++ b/docs/activity-outcome-rejection-audit.md @@ -23,6 +23,10 @@ outcome merely because its response was retried. - Authenticate and enforce namespace, task, attempt and original owner binding before any audit write. Unrelated or unknown identities create no record. + Attempt identifiers require exact stored spelling even when the database + collation considers a case alias equal. A missing stored original lease owner + is unavailable evidence; do not manufacture an owner or append a fact without + that binding. These internal checks do not replace HTTP authentication. - Recheck lost authority under the existing storage transaction and locks. Select the reason from stored state and original deadlines. A request cannot nominate its own audit reason or make a valid claim stale. diff --git a/src/V2/Support/ActivityOutcomeRejection.php b/src/V2/Support/ActivityOutcomeRejection.php index cbb693bc9..90c07c938 100644 --- a/src/V2/Support/ActivityOutcomeRejection.php +++ b/src/V2/Support/ActivityOutcomeRejection.php @@ -31,6 +31,9 @@ public static function record( } $attempt = ActivityAttempt::query()->lockForUpdate()->find($attemptId); if (! $attempt instanceof ActivityAttempt + || $attempt->id !== $attemptId + || ! is_string($attempt->lease_owner) + || $attempt->lease_owner === '' || $attempt->workflow_task_id !== $task->id || $attempt->activity_execution_id !== $execution->id || $attempt->workflow_run_id !== $run->id diff --git a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php index a323f59f6..44aac5194 100644 --- a/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php +++ b/tests/Unit/V2/ActivityOutcomeRejectionAuditTest.php @@ -25,8 +25,8 @@ use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Models\WorkflowTimelineEntry; -use Workflow\V2\Support\ActivityTimeoutEnforcer; use Workflow\V2\Support\ActivityOutcomeRejection; +use Workflow\V2\Support\ActivityTimeoutEnforcer; use Workflow\V2\Support\HistoryBudget; use Workflow\V2\Support\RunTimelineProjector; use Workflow\V2\Support\SqliteTransaction; From bc421416489775f651c748b5da61269d7a8ea5bf Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:13:15 -0600 Subject: [PATCH 10/12] fix: audit outcomes refused by the early expired-attempt fence (#878) Main added refusedExpiredAttempt (dec2aff3), which refuses an expired or settled callback under the run lock before the task and execution locks. After merging main, a lost outcome returned there and never reached the audit append, so no ActivityOutcomeRejected fact was written. Pass the attempted outcome into that path and append the audit fact under the run lock. The attempt row lock follows the run lock, the order retention uses, so no reverse lock dependency is added. The retention test now allows exactly that one payload-free audit append. Co-Authored-By: Claude Opus 5.5 --- src/V2/Support/ActivityOutcomeRecorder.php | 47 +++++++++++++++---- .../Unit/V2/ActivityOutcomeRetentionTest.php | 17 ++++++- 2 files changed, 54 insertions(+), 10 deletions(-) diff --git a/src/V2/Support/ActivityOutcomeRecorder.php b/src/V2/Support/ActivityOutcomeRecorder.php index cf2a63738..559019cbb 100644 --- a/src/V2/Support/ActivityOutcomeRecorder.php +++ b/src/V2/Support/ActivityOutcomeRecorder.php @@ -43,8 +43,15 @@ public static function record( ): array { $sqlite = (new ActivityAttempt())->getConnection() ->getDriverName() === 'sqlite'; + $attemptedOutcome = $throwable === null ? 'completed' : 'failed'; if (! $sqlite) { - $refused = self::refusedExpiredAttempt($taskId, $attemptId, $attemptCount, $admissionDeadline); + $refused = self::refusedExpiredAttempt( + $taskId, + $attemptId, + $attemptCount, + $attemptedOutcome, + $admissionDeadline, + ); if ($refused !== null) { return $refused; } @@ -60,12 +67,19 @@ public static function record( $backoffSeconds, $codec, $sqlite, + $attemptedOutcome, $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); + $refused = self::refusedExpiredAttempt( + $taskId, + $attemptId, + $attemptCount, + $attemptedOutcome, + $admissionDeadline, + ); if ($refused !== null) { return $refused; } @@ -109,7 +123,6 @@ public static function record( ->lockForUpdate() ->findOrFail($lockedExecution->workflow_run_id); - $attemptedOutcome = $throwable === null ? 'completed' : 'failed'; $abandoned = ActivityAbandonment::allows($run, $lockedExecution); if (in_array($run->status, [RunStatus::Cancelled, RunStatus::Terminated], true) && ! $abandoned) { $reason = $run->status === RunStatus::Terminated @@ -582,6 +595,7 @@ private static function refusedExpiredAttempt( string $taskId, string $attemptId, int $attemptNumber, + string $attemptedOutcome, ?int $admissionDeadline, ): ?array { $original = ActivityAttempt::query()->find($attemptId); @@ -598,8 +612,8 @@ private static function refusedExpiredAttempt( $connection = $original->getConnection(); $runId = (string) $original->workflow_run_id; - return SqliteTransaction::run(static function () use ($original): ?array { - return self::refusedExpiredAttemptUnderRunLock($original); + return SqliteTransaction::run(static function () use ($original, $attemptedOutcome): ?array { + return self::refusedExpiredAttemptUnderRunLock($original, $attemptedOutcome); }, admission: static function () use ($connection, $runId): void { WorkflowRun::on($connection->getName())->whereKey($runId)->toBase()->update([ 'id' => $runId, @@ -610,8 +624,10 @@ private static function refusedExpiredAttempt( /** * @return array{recorded: bool, reason: string|null, next_task: WorkflowTask|null}|null */ - private static function refusedExpiredAttemptUnderRunLock(ActivityAttempt $original): ?array - { + private static function refusedExpiredAttemptUnderRunLock( + ActivityAttempt $original, + string $attemptedOutcome, + ): ?array { $connection = $original->getConnectionName(); $run = WorkflowRun::on($connection)->whereKey($original->workflow_run_id)->lockForUpdate()->first(); if (! $run instanceof WorkflowRun || $run->details_pruned_at !== null) { @@ -631,8 +647,23 @@ private static function refusedExpiredAttemptUnderRunLock(ActivityAttempt $origi } $reason = self::expiredAttemptReason($run, $execution, $task, $attempt); + if ($reason === null) { + return null; + } + + // The attempt row lock taken here follows the run lock, the same order + // retention uses, so the audit append adds no reverse lock dependency. + ActivityOutcomeRejection::record( + $run, + $execution, + $task, + $attempt->id, + $attempt->attempt_number, + $attemptedOutcome, + $reason, + ); - return $reason === null ? null : self::ignored($reason); + return self::ignored($reason); } private static function expiredAttemptReason( diff --git a/tests/Unit/V2/ActivityOutcomeRetentionTest.php b/tests/Unit/V2/ActivityOutcomeRetentionTest.php index 8c93faacc..36b571190 100644 --- a/tests/Unit/V2/ActivityOutcomeRetentionTest.php +++ b/tests/Unit/V2/ActivityOutcomeRetentionTest.php @@ -16,6 +16,7 @@ use Workflow\V2\Contracts\ActivityLeaseRenewal; use Workflow\V2\Contracts\ActivityTaskBridge; use Workflow\V2\Enums\ActivityAttemptStatus; +use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; @@ -102,7 +103,7 @@ public function testPruningAndRefusedOutcomeCannotDeadlockOrRestoreDetails( self::assertNull(WorkflowRunSummary::query()->find($run->id)); } - public function testRefusedExpiredOutcomeDoesNotPreparePayloadOrWriteHistory(): void + public function testRefusedExpiredOutcomeDoesNotPreparePayloadAndOnlyAppendsItsAudit(): void { [$bridge, $claim, $run] = $this->claimedOriginal(); $this->travel(301) @@ -120,7 +121,19 @@ public function __serialize(): array self::assertFalse($reply['recorded']); self::assertSame('lease_expired', $reply['reason']); - self::assertSame($before, $run->historyEvents()->get()->map->getAttributes()->all()); + $after = $run->historyEvents() + ->orderBy('sequence') + ->get(); + // The existing history is untouched. The only append is the bounded + // diagnostic audit fact, which carries no result payload (#878). + self::assertSame($before, $after->take(count($before))->map->getAttributes()->values()->all()); + self::assertCount(count($before) + 1, $after); + $audit = $after->last(); + self::assertSame(HistoryEventType::ActivityOutcomeRejected, $audit->event_type); + self::assertSame('completed', $audit->payload['outcome']); + self::assertSame('lease_expired', $audit->payload['reason']); + self::assertSame($claim['activity_attempt_id'], $audit->payload['activity_attempt_id']); + self::assertArrayNotHasKey('result', $audit->payload); self::assertSame( ActivityAttemptStatus::Running, ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status From a4600dec046a5bf16f9ced0688c3cf8b47cad34a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:30:39 -0600 Subject: [PATCH 11/12] fix: read the attempt without a lock in the outcome audit (#878) Lease renewal locks attempt, execution, run, then task. The audit append ran under the run lock and then asked for the attempt lock, so a late outcome and a renewal of the same attempt could deadlock. Every caller already holds the run lock, which serializes audit appends and their duplicate check, so the attempt is now read without a lock. Add a deterministic PostgreSQL/MySQL interleaving: renewal holds the attempt and waits for the run, while the late completion holds the run and appends its audit. With the old attempt lock it deadlocks on both databases. Add replay corpus evidence: an expired-lease late completion appends the recorded ActivityOutcomeRejected fact, and that history cold-replays as a waiting run. Co-Authored-By: Claude Opus 5.5 --- src/V2/Support/ActivityOutcomeRecorder.php | 4 +- src/V2/Support/ActivityOutcomeRejection.php | 10 +- .../V2EmbeddedReplayRegressionCorpusTest.php | 49 +++++++++ ...y-lease-expired-late-completion-audit.json | 64 +++++++++++ .../ActivityOutcomeRetentionProcessTest.php | 53 +++++++++ .../Unit/V2/ActivityOutcomeRetentionTest.php | 102 ++++++++++++++++++ 6 files changed, 278 insertions(+), 4 deletions(-) create mode 100644 tests/Fixtures/V2/ReplayRegression/activity-lease-expired-late-completion-audit.json diff --git a/src/V2/Support/ActivityOutcomeRecorder.php b/src/V2/Support/ActivityOutcomeRecorder.php index 559019cbb..692cf0cc4 100644 --- a/src/V2/Support/ActivityOutcomeRecorder.php +++ b/src/V2/Support/ActivityOutcomeRecorder.php @@ -651,8 +651,8 @@ private static function refusedExpiredAttemptUnderRunLock( return null; } - // The attempt row lock taken here follows the run lock, the same order - // retention uses, so the audit append adds no reverse lock dependency. + // The audit append reads the attempt without a lock, so it adds no lock + // after this run lock (renewal locks the attempt before the run). ActivityOutcomeRejection::record( $run, $execution, diff --git a/src/V2/Support/ActivityOutcomeRejection.php b/src/V2/Support/ActivityOutcomeRejection.php index 90c07c938..ffd2df018 100644 --- a/src/V2/Support/ActivityOutcomeRejection.php +++ b/src/V2/Support/ActivityOutcomeRejection.php @@ -15,7 +15,13 @@ final class ActivityOutcomeRejection { /** - * Record only under the original task/execution/run mutation locks. + * Record only while holding the run lock. Every caller holds it, so audit + * appends and their duplicate check are serialized per run. + * + * The attempt row is read without a lock. Lease renewal locks the attempt + * before the run, so asking for the attempt lock here, after the run lock, + * could deadlock with a renewal. A renewal that has not committed cannot + * change what this refusal observed. */ public static function record( WorkflowRun $run, @@ -29,7 +35,7 @@ public static function record( if ($run->details_pruned_at !== null || ! in_array($outcome, ['completed', 'failed'], true)) { return null; } - $attempt = ActivityAttempt::query()->lockForUpdate()->find($attemptId); + $attempt = ActivityAttempt::query()->find($attemptId); if (! $attempt instanceof ActivityAttempt || $attempt->id !== $attemptId || ! is_string($attempt->lease_owner) diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index aaed0f209..98ee3ca85 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -959,6 +959,12 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void continue; } + if ($fixture['id'] === 'activity-lease-expired-late-completion-audit') { + $this->clearWorkflowState(); + $this->assertExpiredLeaseLateCompletionAppendsItsAudit($fixture); + continue; + } + if ($fixture['id'] === 'external-worker-pending-count-limit-rollback') { $this->assertExternalWorkerPendingCountLimitRollsBackCompletion($fixture); continue; @@ -1395,6 +1401,49 @@ private function assertUpdateTailCheckpointRetainsItsCursor(array $fixture): voi /** * @param array $fixture */ + /** + * #878: a completion that arrives after its lease expired is refused by the + * early expired-attempt fence. That refusal must still append the bounded + * audit fact this fixture records, and nothing else. + * + * @param array $fixture + */ + private function assertExpiredLeaseLateCompletionAppendsItsAudit(array $fixture): void + { + $expected = collect($fixture['history'])->firstWhere('event_type', 'ActivityOutcomeRejected')['payload']; + $workflow = \Workflow\V2\WorkflowStub::make($fixture['workflow']['type'], 'late-completion-audit'); + $workflow->start(...$fixture['workflow']['arguments']); + $task = \Workflow\V2\Models\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); + $claim = $bridge->claimStatus($bridge->poll(null, null, 1)[0]['task_id'], $expected['lease_owner']); + $this->assertTrue($claim['claimed']); + $run = WorkflowRun::query()->findOrFail($workflow->runId()); + $before = $run->historyEvents() + ->count(); + + $this->travel(301) + ->seconds(); + $reply = $bridge->complete($claim['activity_attempt_id'], 'late result'); + + $this->assertFalse($reply['recorded']); + $this->assertSame($expected['reason'], $reply['reason']); + $events = $run->historyEvents() + ->orderBy('sequence') + ->get(); + $this->assertCount($before + 1, $events); + $audit = $events->last(); + $this->assertSame(HistoryEventType::ActivityOutcomeRejected, $audit->event_type); + $this->assertSame($claim['activity_attempt_id'], $audit->payload['activity_attempt_id']); + $this->assertSame(array_keys($expected), array_keys($audit->payload)); + foreach (['attempt_number', 'lease_owner', 'outcome', 'reason'] as $key) { + $this->assertSame($expected[$key], $audit->payload[$key], $key); + } + } + private function assertExternalWorkerPendingCountLimitRollsBackCompletion(array $fixture): void { $cases = $fixture['external_worker_pending_counts']; diff --git a/tests/Fixtures/V2/ReplayRegression/activity-lease-expired-late-completion-audit.json b/tests/Fixtures/V2/ReplayRegression/activity-lease-expired-late-completion-audit.json new file mode 100644 index 000000000..67206dd6c --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/activity-lease-expired-late-completion-audit.json @@ -0,0 +1,64 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "activity-lease-expired-late-completion-audit", + "protocol_version": "1.0", + "bindings": ["php"], + "consumers": ["workflow-fiber-runner"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRetryWorkflow", + "arguments": ["Ada"], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-03T12:00:00.000000Z" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "recorded_at": "2026-11-03T12:00:00.000001Z", + "payload": { + "sequence": 1, + "activity_execution_id": "01j00000000000000000000901", + "activity_class": "Tests\\Fixtures\\V2\\TestRetryActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestRetryActivity" + } + }, + { + "sequence": 3, + "event_type": "ActivityStarted", + "recorded_at": "2026-11-03T12:00:00.000002Z", + "payload": { + "sequence": 1, + "activity_execution_id": "01j00000000000000000000901", + "activity_attempt_id": "01j0000000000000000000090a", + "activity_class": "Tests\\Fixtures\\V2\\TestRetryActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestRetryActivity", + "attempt_number": 1 + } + }, + { + "sequence": 4, + "event_type": "ActivityOutcomeRejected", + "recorded_at": "2026-11-03T12:05:01.000000Z", + "payload": { + "activity_execution_id": "01j00000000000000000000901", + "activity_attempt_id": "01j0000000000000000000090a", + "attempt_number": 1, + "lease_owner": "original-worker", + "outcome": "completed", + "reason": "lease_expired", + "observed_at": "2026-11-03T12:05:01.000000Z" + } + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [] + } +} diff --git a/tests/Support/ActivityOutcomeRetentionProcessTest.php b/tests/Support/ActivityOutcomeRetentionProcessTest.php index 26b1ed7e1..c55a70cf9 100644 --- a/tests/Support/ActivityOutcomeRetentionProcessTest.php +++ b/tests/Support/ActivityOutcomeRetentionProcessTest.php @@ -41,6 +41,49 @@ public function testOutcomeAndRetention(): void 'recorded' => false, 'reason' => 'pruned', ]; + } elseif ($input['outcome'] === 'renew') { + // Hold the attempt lock, as renewal takes it first, until released. + // Then announce the run lock request that waits on the audit. + $paused = false; + $connection->listen(static function (QueryExecuted $event) use ($input, &$paused): void { + if ($paused || ! str_contains(strtolower($event->sql), 'for update') + || ! str_contains($event->sql, 'activity_attempts')) { + return; + } + $paused = true; + file_put_contents($input['ready'], 'attempt locked'); + self::waitFor($input['continue'], $input['deadline']); + }); + $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['waiting'], 'requesting run lock'); + } + }); + $status = $this->app->make(ActivityTaskBridge::class)->renewLease( + $input['attempt_id'], + $input['lease_owner'], + ); + $reply = [ + 'recorded' => $status['lease_renewed'], + 'reason' => $status['reason'], + ]; + } elseif ($input['pause_after_run_lock'] ?? false) { + // Hold the run lock before the audit append until released. + $paused = false; + $connection->listen(static function (QueryExecuted $event) use ($input, &$paused): void { + if ($paused || ! str_contains(strtolower($event->sql), 'for update') + || ! str_contains($event->sql, 'workflow_runs')) { + return; + } + $paused = true; + file_put_contents($input['ready'], 'run locked'); + self::waitFor($input['continue'], $input['deadline']); + }); + $reply = $this->app->make(ActivityTaskBridge::class)->complete( + $input['attempt_id'], + 'discarded late result', + ); + self::assertFalse($reply['recorded']); } else { // Signal before this query can wait on retention's run lock. The // preceding implementation already holds task/execution locks here. @@ -94,6 +137,16 @@ protected function defineEnvironment($app): void $app['config']->set('logging.default', 'null'); } + private static function waitFor(string $path, int $deadline): void + { + while (! is_file($path)) { + if (hrtime(true) >= $deadline) { + self::fail('The interleaving exceeded its original process budget.'); + } + usleep(1_000); + } + } + private function input(): array { $path = getenv('DWF_RETENTION_PROCESS_INPUT'); diff --git a/tests/Unit/V2/ActivityOutcomeRetentionTest.php b/tests/Unit/V2/ActivityOutcomeRetentionTest.php index 36b571190..30d573e33 100644 --- a/tests/Unit/V2/ActivityOutcomeRetentionTest.php +++ b/tests/Unit/V2/ActivityOutcomeRetentionTest.php @@ -292,6 +292,88 @@ public function testSqliteRefusalMetadataKeepsTheOriginalPublicationBudget(): vo ); } + public function testLeaseRenewalAndLateOutcomeAuditCannotDeadlock(): 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(); + $directory = sys_get_temp_dir() . '/workflow-renewal-audit-' . bin2hex(random_bytes(8)); + mkdir($directory, 0700); + $workers = []; + try { + $input = static fn (int $index, array $extra): array => array_merge([ + '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'], + 'lease_owner' => $claim['lease_owner'], + 'direct' => false, + 'observed_at' => now() + ->toISOString(), + 'deadline' => $deadline, + 'ready' => $directory . '/ready-' . $index, + 'continue' => $directory . '/continue-' . $index, + 'waiting' => $directory . '/waiting-' . $index, + 'report' => $directory . '/report-' . $index . '.json', + ], $extra); + + // Renewal locks the attempt first and holds it. + $workers[] = $this->startProcessWorker($directory, 0, $input(0, [ + 'outcome' => 'renew', + ])); + $this->awaitFile($directory . '/ready-0', $workers, $deadline); + // The late completion takes the run lock and holds it before its audit. + $workers[] = $this->startProcessWorker($directory, 1, $input(1, [ + 'outcome' => 'completed', + 'pause_after_run_lock' => true, + ])); + $this->awaitFile($directory . '/ready-1', $workers, $deadline); + // Renewal now waits for the run lock while holding the attempt. + file_put_contents($directory . '/continue-0', 'continue'); + $this->awaitFile($directory . '/waiting-0', $workers, $deadline); + usleep(100_000); + // The audit append must not wait for the attempt the renewal holds. + file_put_contents($directory . '/continue-1', 'continue'); + while ($workers[0]->isRunning() || $workers[1]->isRunning()) { + if (hrtime(true) >= $deadline) { + self::fail('The original renewal interleaving budget expired.'); + } + usleep(1_000); + } + foreach ($workers as $worker) { + self::assertSame( + 0, + $worker->getExitCode(), + substr($worker->getErrorOutput() . $worker->getOutput(), -8_192) + ); + } + $renewal = json_decode(file_get_contents($directory . '/report-0.json'), true, flags: JSON_THROW_ON_ERROR); + $outcome = 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); + } + + self::assertFalse($renewal['recorded']); + self::assertFalse($outcome['recorded']); + self::assertSame('lease_expired', $outcome['reason']); + $audits = $run->historyEvents() + ->where('event_type', HistoryEventType::ActivityOutcomeRejected) + ->get(); + self::assertCount(1, $audits); + self::assertSame('lease_expired', $audits->sole()->payload['reason']); + } + private function claimedOriginal(): array { $workflow = WorkflowStub::make(TestRetryWorkflow::class, 'outcome-retention'); @@ -309,6 +391,26 @@ private function claimedOriginal(): array return [$bridge, $claim, WorkflowRun::query()->findOrFail($workflow->runId())]; } + private function startProcessWorker(string $directory, int $index, array $input): Process + { + $path = $directory . '/input-' . $index . '.json'; + file_put_contents($path, json_encode($input, 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); + $worker->start(); + + return $worker; + } + private function interleavePruning( WorkflowRun $run, array $claim, From 31d58de736c57bd24e9575fc61d3dbf159668049 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:49:12 -0600 Subject: [PATCH 12/12] test: compare the audit fixture's key set without relying on JSON key order (#878) MySQL returns JSON object keys in its own order, so the corpus hook now compares sorted key sets. The per-field assertions are unchanged. Co-Authored-By: Claude Opus 5.5 --- tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 98ee3ca85..16f3fc96b 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -1438,7 +1438,12 @@ private function assertExpiredLeaseLateCompletionAppendsItsAudit(array $fixture) $audit = $events->last(); $this->assertSame(HistoryEventType::ActivityOutcomeRejected, $audit->event_type); $this->assertSame($claim['activity_attempt_id'], $audit->payload['activity_attempt_id']); - $this->assertSame(array_keys($expected), array_keys($audit->payload)); + // JSON columns do not keep key order on every database (MySQL reorders). + $expectedKeys = array_keys($expected); + $actualKeys = array_keys($audit->payload); + sort($expectedKeys); + sort($actualKeys); + $this->assertSame($expectedKeys, $actualKeys); foreach (['attempt_number', 'lease_owner', 'outcome', 'reason'] as $key) { $this->assertSame($expected[$key], $audit->payload[$key], $key); }