fix(temporal): an activity failure reports the attempt the server ran (#547)

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

src/Bridge/Temporal/Worker/TemporalExecutionHistory.php

@@ -67,6 +67,12 @@ final class TemporalExecutionHistory implements WorkflowHistorySourceInterface
     /** @var array<int, string> scheduled event ID → activity ID */
     private array $scheduledEventIdToActivityId = [];
 
+    /**
+     * @var array<int, int> 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<string, mixed> 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,15 @@ 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 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
+    {
+        return max(1, $this->startedEventIdToAttempt[$startedEventId] ?? 1);
+    }
 }

tests/unit/Bridge/Temporal/Worker/AnActivityFailureReportsItsAttemptTest.php

@@ -0,0 +1,100 @@
+<?php
+
+declare(strict_types=1);
+
+namespace unit\Gplanchat\Bridge\Temporal\Worker;
+
+use Gplanchat\Bridge\Temporal\Worker\TemporalExecutionHistory;
+use Gplanchat\Durable\Event\ActivityFailed;
+use Gplanchat\Durable\Exception\DurableActivityFailedException;
+use Gplanchat\Durable\Failure\ActivityFailureEventFactory;
+use Gplanchat\Durable\Failure\ActivityRetryState;
+use PHPUnit\Framework\TestCase;
+use Temporal\Api\Common\V1\ActivityType;
+use Temporal\Api\Enums\V1\EventType;
+use Temporal\Api\Enums\V1\TimeoutType;
+use Temporal\Api\Failure\V1\ApplicationFailureInfo;
+use Temporal\Api\Failure\V1\Failure;
+use Temporal\Api\Failure\V1\TimeoutFailureInfo;
+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\HistoryEvent;
+
+/**
+ * A workflow reading `$e->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;
+    }
+}

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<string, mixed> $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;
+    }
 }