From 9d11935def719ed168767606712131d44b7d0841 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gr=C3=A9gory=20Planchat?= Date: Fri, 25 Sep 2026 23:40:42 +0200 Subject: [PATCH 1/3] fix(temporal): an activity failure reports the attempt the server ran (#547) A failure or a timeout read on Temporal always said attempt 1, where the journal backends report the attempt that failed, so $e->attempt() and the exception's message depended on the backend. The history now records each ActivityTaskStarted's attempt, and ACTIVITY_TASK_FAILED and ACTIVITY_TASK_TIMED_OUT read it through their started_event_id. With no started event (a schedule-to-start timeout) it stays 1, as on the journal. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Worker/TemporalExecutionHistory.php | 26 ++++- ...AnActivityFailureReportsItsAttemptTest.php | 100 ++++++++++++++++++ 2 files changed, 124 insertions(+), 2 deletions(-) create mode 100644 tests/unit/Bridge/Temporal/Worker/AnActivityFailureReportsItsAttemptTest.php diff --git a/src/Bridge/Temporal/Worker/TemporalExecutionHistory.php b/src/Bridge/Temporal/Worker/TemporalExecutionHistory.php index 51985f3d..a3a1b5cb 100644 --- a/src/Bridge/Temporal/Worker/TemporalExecutionHistory.php +++ b/src/Bridge/Temporal/Worker/TemporalExecutionHistory.php @@ -67,6 +67,12 @@ final class TemporalExecutionHistory implements WorkflowHistorySourceInterface /** @var array scheduled event ID → activity ID */ private array $scheduledEventIdToActivityId = []; + /** + * @var array started event ID → the attempt it started; Temporal writes only the last + * attempt's ActivityTaskStarted, and a failure or a timeout points at it (#547) + */ + private array $startedEventIdToAttempt = []; + /** @var array activity ID → result (for completed activities) */ private array $activityResults = []; @@ -304,6 +310,13 @@ private function consumeEvent(HistoryEvent $event): void // journal backends raise a timeout as a RuntimeException naming it, so a workflow // catches the same failure on every backend. Unread, it left the slot empty and the // workflow waiting forever (#544). + case EventType::EVENT_TYPE_ACTIVITY_TASK_STARTED: + $attr = $event->getActivityTaskStartedEventAttributes(); + if (null !== $attr) { + $this->startedEventIdToAttempt[$eventId] = $attr->getAttempt(); + } + break; + case EventType::EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT: $attr = $event->getActivityTaskTimedOutEventAttributes(); $activityId = null !== $attr ? $this->scheduledEventIdToActivityId[$attr->getScheduledEventId()] ?? null : null; @@ -319,7 +332,7 @@ private function consumeEvent(HistoryEvent $event): void $this->activityFailures[$activityId] = new DurableActivityFailedException( $activityId, $this->activityNames[$activityId] ?? '', - 1, + $this->attemptOf($attr?->getStartedEventId() ?? 0), new FailureEnvelope(\RuntimeException::class, \sprintf('Activity %stimeout exceeded.', $timeout), 0, [], null, []), ); } @@ -341,7 +354,7 @@ private function consumeEvent(HistoryEvent $event): void $this->activityFailures[$activityId] = new DurableActivityFailedException( $activityId, $this->activityNames[$activityId] ?? '', - 1, + $this->attemptOf($attr->getStartedEventId()), new FailureEnvelope( \is_string($type) && '' !== $type ? $type : \RuntimeException::class, $message, @@ -888,4 +901,13 @@ public function scheduledEventIdForNexusOperation(string $operationId): ?int { return $this->nexusOperationToScheduledEventId[$operationId] ?? null; } + + /** + * The attempt a failure or a timeout ended, read from the ActivityTaskStarted it points at. With + * none (a schedule-to-start timeout, before any start), the first, as the journal backends say. + */ + private function attemptOf(int $startedEventId): int + { + return max(1, $this->startedEventIdToAttempt[$startedEventId] ?? 1); + } } diff --git a/tests/unit/Bridge/Temporal/Worker/AnActivityFailureReportsItsAttemptTest.php b/tests/unit/Bridge/Temporal/Worker/AnActivityFailureReportsItsAttemptTest.php new file mode 100644 index 00000000..47f2a5b4 --- /dev/null +++ b/tests/unit/Bridge/Temporal/Worker/AnActivityFailureReportsItsAttemptTest.php @@ -0,0 +1,100 @@ +attempt()` gets the attempt that failed, on Temporal as on the journal + * backends. Temporal writes only the last attempt's ActivityTaskStarted, and the failure points at + * it (#547). + * + * @internal + */ +final class AnActivityFailureReportsItsAttemptTest extends TestCase +{ + public function testAFailureOnTheThirdAttemptReportsTheAttemptTheJournalReports(): void + { + $failed = new HistoryEvent(['event_id' => 4, 'event_type' => EventType::EVENT_TYPE_ACTIVITY_TASK_FAILED]); + $failed->setActivityTaskFailedEventAttributes(new ActivityTaskFailedEventAttributes([ + 'scheduled_event_id' => 2, + 'started_event_id' => 3, + 'failure' => new Failure(['message' => 'card declined', 'application_failure_info' => new ApplicationFailureInfo(['type' => \RuntimeException::class])]), + ])); + + $onTemporal = self::failure(3, $failed); + $journal = ActivityFailureEventFactory::fromActivityThrowable('exec-1', 'act-1', 'Charge', 3, new \RuntimeException('card declined'), ActivityRetryState::MaximumAttemptsReached); + self::assertInstanceOf(ActivityFailed::class, $journal); + $onTheJournal = DurableActivityFailedException::toThrowable($journal); + self::assertInstanceOf(DurableActivityFailedException::class, $onTheJournal); + + self::assertSame(3, $onTemporal->attempt()); + self::assertSame($onTheJournal->attempt(), $onTemporal->attempt()); + self::assertSame($onTheJournal->getMessage(), $onTemporal->getMessage(), 'the attempt is part of the message a workflow logs'); + } + + public function testATimeoutAfterItsSecondAttemptReportsTheSecondAttempt(): void + { + self::assertSame(2, self::failure(2, self::timedOut(TimeoutType::TIMEOUT_TYPE_START_TO_CLOSE, 3))->attempt()); + } + + public function testATimeoutBeforeAnyStartReportsTheFirstAttemptAsTheJournalDoes(): void + { + // Schedule-to-start: no attempt ever started, and the journal reports the first one. + self::assertSame(1, self::failure(null, self::timedOut(TimeoutType::TIMEOUT_TYPE_SCHEDULE_TO_START, 0))->attempt()); + } + + private static function timedOut(int $timeoutType, int $startedEventId): HistoryEvent + { + $event = new HistoryEvent(['event_id' => 4, 'event_type' => EventType::EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT]); + $event->setActivityTaskTimedOutEventAttributes(new ActivityTaskTimedOutEventAttributes([ + 'scheduled_event_id' => 2, + 'started_event_id' => $startedEventId, + 'failure' => new Failure(['timeout_failure_info' => new TimeoutFailureInfo(['timeout_type' => $timeoutType])]), + ])); + + return $event; + } + + /** + * @param int|null $startedAttempt the attempt of the last ActivityTaskStarted, none when null + */ + private static function failure(?int $startedAttempt, HistoryEvent $ending): DurableActivityFailedException + { + $scheduled = new HistoryEvent(['event_id' => 2, 'event_type' => EventType::EVENT_TYPE_ACTIVITY_TASK_SCHEDULED]); + $scheduled->setActivityTaskScheduledEventAttributes(new ActivityTaskScheduledEventAttributes([ + 'activity_id' => 'act-1', + 'activity_type' => new ActivityType(['name' => 'Charge']), + ])); + $events = [new HistoryEvent(['event_id' => 1, 'event_type' => EventType::EVENT_TYPE_WORKFLOW_EXECUTION_STARTED]), $scheduled]; + if (null !== $startedAttempt) { + $started = new HistoryEvent(['event_id' => 3, 'event_type' => EventType::EVENT_TYPE_ACTIVITY_TASK_STARTED]); + $started->setActivityTaskStartedEventAttributes(new ActivityTaskStartedEventAttributes(['scheduled_event_id' => 2, 'attempt' => $startedAttempt])); + $events[] = $started; + } + $events[] = $ending; + + $failed = TemporalExecutionHistory::fromEvents($events)->findActivitySlotResult(0)['failed'] ?? null; + self::assertInstanceOf(DurableActivityFailedException::class, $failed); + + return $failed; + } +} From 3f98f86b71c4fe7f937b8001ddf3f745c63559be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gr=C3=A9gory=20Planchat?= Date: Fri, 25 Sep 2026 23:46:57 +0200 Subject: [PATCH 2/3] test(temporal): what the real attempt does to a replay, both ways (#547, review) Three runner tests over a history where double(2) ran three attempts, failed, and the workflow scheduled greet(...) under the old reading: - copying $e->attempt() into greet's payload diverges on replay (it passes with main's reader, which still says attempt 1); - a workflow that does not copy it replays with no divergence; - UPGRADE.md's way out, a change point that keeps attempt 1 for runs that started before it, replays with no divergence. The attempt docblock also names the schedule-to-close case that reads 1. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Worker/TemporalExecutionHistory.php | 4 +- .../Worker/WorkflowTaskRunnerTest.php | 96 +++++++++++++++++++ 2 files changed, 99 insertions(+), 1 deletion(-) diff --git a/src/Bridge/Temporal/Worker/TemporalExecutionHistory.php b/src/Bridge/Temporal/Worker/TemporalExecutionHistory.php index a3a1b5cb..219d32ac 100644 --- a/src/Bridge/Temporal/Worker/TemporalExecutionHistory.php +++ b/src/Bridge/Temporal/Worker/TemporalExecutionHistory.php @@ -904,7 +904,9 @@ public function scheduledEventIdForNexusOperation(string $operationId): ?int /** * The attempt a failure or a timeout ended, read from the ActivityTaskStarted it points at. With - * none (a schedule-to-start timeout, before any start), the first, as the journal backends say. + * none it is 1: a schedule-to-start timeout before any start, as the journal backends say, but + * also a schedule-to-close timeout that hits while a later attempt is still queued, since + * ACTIVITY_TASK_TIMED_OUT carries no attempt of its own. */ private function attemptOf(int $startedEventId): int { diff --git a/tests/unit/Bridge/Temporal/Worker/WorkflowTaskRunnerTest.php b/tests/unit/Bridge/Temporal/Worker/WorkflowTaskRunnerTest.php index 834a9d29..3d6f8b45 100644 --- a/tests/unit/Bridge/Temporal/Worker/WorkflowTaskRunnerTest.php +++ b/tests/unit/Bridge/Temporal/Worker/WorkflowTaskRunnerTest.php @@ -12,10 +12,14 @@ use Gplanchat\Bridge\Temporal\WorkflowServiceClientInterface; use Gplanchat\Durable\Duration; use Gplanchat\Durable\Exception\DeadlineExceededException; +use Gplanchat\Durable\Exception\DurableActivityFailedException; +use Gplanchat\Durable\Exception\WorkflowTaskFailure; +use Gplanchat\Durable\Versioning\ChangePoint; use Gplanchat\Durable\WorkflowEnvironment; use Gplanchat\Durable\WorkflowRegistry; use PHPUnit\Framework\MockObject\MockObject; use PHPUnit\Framework\TestCase; +use Temporal\Api\Common\V1\ActivityType; use Temporal\Api\Common\V1\Payloads; use Temporal\Api\Common\V1\WorkflowExecution; use Temporal\Api\Common\V1\WorkflowType; @@ -27,6 +31,7 @@ use Temporal\Api\History\V1\ActivityTaskCompletedEventAttributes; use Temporal\Api\History\V1\ActivityTaskFailedEventAttributes; use Temporal\Api\History\V1\ActivityTaskScheduledEventAttributes; +use Temporal\Api\History\V1\ActivityTaskStartedEventAttributes; use Temporal\Api\History\V1\ActivityTaskTimedOutEventAttributes; use Temporal\Api\History\V1\History; use Temporal\Api\History\V1\HistoryEvent; @@ -675,4 +680,95 @@ public function testSignalNotYetReceivedSuspendsWorkflow(): void // Workflow is suspended waiting for signal → no commands emitted self::assertEmpty($result->commands, 'No commands when workflow is suspended waiting for signal'); } + + /** + * #547: the attempt a failed activity reports is now the one the server ran. A workflow that + * copies it into a later payload, in a run recorded under the old reading (attempt 1), diverges + * on replay; UPGRADE.md says to drain such runs or branch on versionForChangeId. Kept to pin + * that the divergence is loud, not a silently different payload. + */ + public function testAnAttemptCopiedIntoAPayloadDivergesOnARunRecordedUnderTheOldReading(): void + { + $this->expectException(WorkflowTaskFailure::class); + $this->expectExceptionMessageMatches('/Replay divergence at activity slot 1 .*attempt-1.*attempt-3/s'); + + $this->replayRetriedThenFailed('attempt-1', static fn(DurableActivityFailedException $e, WorkflowEnvironment $env): string => 'attempt-' . $e->attempt()); + } + + /** + * #547: a workflow that does not put the attempt into a payload replays such a run unchanged. + */ + public function testAPlainReplayOfARetriedThenFailedActivityDoesNotDiverge(): void + { + $result = $this->replayRetriedThenFailed('x', static fn(DurableActivityFailedException $e, WorkflowEnvironment $env): string => 'x'); + + self::assertSame([], array_map(static fn($c): int => $c->getCommandType(), array_filter( + $result->commands, + static fn($c): bool => CommandType::COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION === $c->getCommandType(), + ))); + } + + /** + * #547: UPGRADE.md's way out for such a run, a change point that keeps the old reading for an + * execution that started before it, replays the old history without diverging. + */ + public function testTheVersionedWayOutReplaysARunRecordedUnderTheOldReading(): void + { + $result = $this->replayRetriedThenFailed('attempt-1', static fn(DurableActivityFailedException $e, WorkflowEnvironment $env): string => 'attempt-' . ( + ChangePoint::DEFAULT_VERSION === $env->version('real-activity-attempt', ChangePoint::DEFAULT_VERSION, 1) ? 1 : $e->attempt() + )); + + self::assertSame([], array_filter($result->commands, static fn($c): bool => CommandType::COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION === $c->getCommandType())); + } + + /** + * double(2) ran three attempts and failed; the workflow caught it and scheduled greet(...), + * recorded with `$recordedName`. + * + * @param \Closure(DurableActivityFailedException, WorkflowEnvironment): string $name + */ + private function replayRetriedThenFailed(string $recordedName, \Closure $name): \Gplanchat\Bridge\Temporal\Worker\WorkflowTaskResult + { + $registry = new WorkflowRegistry(); + $registry->registerFactory('Probe', static fn(array $payload) => static function (WorkflowEnvironment $env) use ($name): string { + $stub = $env->activityStub(SuiteActivities::class); + + try { + return (string) $env->await($stub->double(2)); + } catch (DurableActivityFailedException $e) { + return $env->await($stub->greet($name($e, $env))); + } + }); + + $started = self::makeEvent(3, EventType::EVENT_TYPE_ACTIVITY_TASK_STARTED); + $started->setActivityTaskStartedEventAttributes(new ActivityTaskStartedEventAttributes(['scheduled_event_id' => 2, 'attempt' => 3])); + $failed = self::makeActivityFailed(4, 2, 'boom'); + $failed->getActivityTaskFailedEventAttributes()?->setStartedEventId(3); + + return $this->makeRunner($registry)->run(self::buildPoll('token-attempt', 'wf-attempt', 'Probe', [ + self::makeStarted(1), + self::scheduledWithInput(2, 'act-1', 'double', ['value' => 2]), + $started, + $failed, + self::scheduledWithInput(5, 'act-2', 'greet', ['name' => $recordedName]), + self::makeEvent(6, EventType::EVENT_TYPE_WORKFLOW_TASK_SCHEDULED), + self::makeEvent(7, EventType::EVENT_TYPE_WORKFLOW_TASK_STARTED), + ])); + } + + /** + * @param array $arguments + */ + private static function scheduledWithInput(int $id, string $activityId, string $type, array $arguments): HistoryEvent + { + $e = self::makeActivityScheduled($id, $activityId); + $attr = $e->getActivityTaskScheduledEventAttributes(); + \assert(null !== $attr); + $attr->setActivityType(new ActivityType(['name' => $type])); + $input = new Payloads(); + $input->setPayloads([JsonPlainPayload::encode(['payload' => $arguments])]); + $attr->setInput($input); + + return $e; + } } From a899ec83b85ba9f252ede15e7fea97901e31070a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gr=C3=A9gory=20Planchat?= Date: Fri, 25 Sep 2026 23:46:57 +0200 Subject: [PATCH 3/3] docs(upgrade): replay, a payload built from $e->attempt() on Temporal (#547, review) In-flight Temporal runs that put the attempt into a later payload diverge on replay once the real attempt is read. The entry says who is affected, the error they would see, and the two ways out: drain, or a change point that keeps attempt 1 for runs that started before it, the snippet a test replays. Co-Authored-By: Claude Opus 5.5 (1M context) --- UPGRADE.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/UPGRADE.md b/UPGRADE.md index 759c1f69..70f713eb 100644 --- a/UPGRADE.md +++ b/UPGRADE.md @@ -664,6 +664,29 @@ label is display text: nothing records it while the run waits, it appears in the uncaught deadline, and replay never compares it. A label on a timer or an activity is refused with an `InvalidArgumentException` (#324). +### Replay: on Temporal, a failed activity reports the attempt the server ran + +**Who is affected**: workflows on the Temporal backend that catch a `DurableActivityFailedException` +and put `$e->attempt()`, or the exception's message (which names the attempt), into the payload of a +later activity, child workflow or Nexus operation. Everyone else has nothing to do. The journal +backends always reported the real attempt; Temporal said `1` (#547). + +A run of that workflow that is in flight when you upgrade was recorded with `attempt 1` in that +payload. Replayed with the new reading, it schedules the real attempt, and the worker refuses the +task: `Replay divergence at activity slot N … history recorded "…attempt-1…", code scheduled +"…attempt-3…"`. Either drain those runs before deploying, or keep the old reading for the runs that +started before the change: + +```php +} catch (DurableActivityFailedException $e) { + $attempt = ChangePoint::DEFAULT_VERSION === $env->version('real-activity-attempt', ChangePoint::DEFAULT_VERSION, 1) + ? 1 // started before the upgrade: the history recorded attempt 1 + : $e->attempt(); + + return $env->await($activities->greet('attempt-' . $attempt)); +} +``` + ## 0.1.0-alpha8 ### The divergence guard compares the payload too