Reduce repeated full-projection lookups during workflow execution

src/V2/Support/IdempotentProjectionUpsert.php

@@ -30,11 +30,33 @@ final class IdempotentProjectionUpsert
      * @param class-string<TModel> $model
      * @param array<string, mixed> $key
      * @param array<string, mixed> $values
+     * @param TModel|null $existing Row loaded by this projection pass, never a cross-task cache.
      * @return TModel
      */
-    public static function upsert(string $model, array $key, array $values): Model
+    public static function upsert(string $model, array $key, array $values, ?Model $existing = null): Model
     {
+        if ($existing !== null) {
+            if (! $existing instanceof $model || ! $existing->exists) {
+                throw new \InvalidArgumentException(
+                    'Prefetched projection must be a persisted instance of the configured model.'
+                );
+            }
+
+            foreach ($key as $column => $value) {
+                if ($existing->getAttribute($column) !== $value) {
+                    throw new \InvalidArgumentException('Prefetched projection must match the upsert key.');
+                }
+            }
+        }
+
         try {
+            if ($existing !== null) {
+                $existing->fill($values)
+                    ->save();
+
+                return $existing;
+            }
+
             /** @var TModel $row */
             $row = $model::query()->updateOrCreate($key, $values);
 

src/V2/Support/RunSummaryProjector.php

@@ -399,9 +399,9 @@ public static function project(WorkflowRun $run): WorkflowRunSummary
             $failureIds,
         );
 
-        RunWaitProjector::project($run);
+        RunWaitProjector::project($run, RunWaitView::forRun($run, $activities, $timers));
         RunTimelineProjector::project($run);
-        RunTimerProjector::project($run);
+        RunTimerProjector::project($run, $timers);
         RunLineageProjector::project($run);
 
         return $summary;

src/V2/Support/RunTimelineProjector.php

@@ -23,6 +23,7 @@ public static function project(WorkflowRun $run, ?array $entries = null): array
     {
         $entries ??= HistoryTimeline::fromHistory($run);
         $entryModel = self::entryModel();
+        $existing = $entryModel::query()->where('workflow_run_id', $run->id)->get()->keyBy('id');
         $seen = [];
         $projected = [];
 
@@ -35,7 +36,14 @@ public static function project(WorkflowRun $run, ?array $entries = null): array
 
             $projectionId = self::projectionId($run->id, $historyEventId);
             $seen[] = $projectionId;
-            $projected[] = self::upsertEntry($run, $entryModel, $projectionId, $historyEventId, $entry);
+            $projected[] = self::upsertEntry(
+                $run,
+                $entryModel,
+                $projectionId,
+                $historyEventId,
+                $entry,
+                $existing->get($projectionId)
+            );
         }
 
         self::historyProjectionMaintenanceRole()
@@ -153,6 +161,7 @@ private static function upsertEntry(
         string $projectionId,
         string $historyEventId,
         array $entry,
+        ?WorkflowTimelineEntry $existing = null,
     ): WorkflowTimelineEntry {
         /** @var WorkflowTimelineEntry $row */
         $row = IdempotentProjectionUpsert::upsert(
@@ -180,6 +189,7 @@ private static function upsertEntry(
                 'failure_id' => self::stringValue($entry['failure_id'] ?? null),
                 'payload' => self::normalizedPayload($entry),
             ],
+            $existing,
         );
 
         return $row;

src/V2/Support/RunTimerProjector.php

@@ -24,6 +24,7 @@ public static function project(WorkflowRun $run, ?array $timers = null): array
         $entryModel = self::entryModel();
         $seen = [];
         $projected = [];
+        $existing = $entryModel::query()->where('workflow_run_id', $run->id)->get()->keyBy('id');
 
         foreach (array_values($timers) as $position => $timer) {
             $timerId = self::stringValue($timer['id'] ?? null);
@@ -64,6 +65,7 @@ public static function project(WorkflowRun $run, ?array $timers = null): array
                     'history_unsupported_reason' => self::stringValue($timer['history_unsupported_reason'] ?? null),
                     'payload' => self::normalizedPayload($timer),
                 ],
+                $existing->get($projectionId),
             );
 
             $projected[] = $row;

src/V2/Support/RunWaitProjector.php

@@ -143,6 +143,7 @@ public static function project(WorkflowRun $run, ?array $waits = null): array
         $waitModel = self::waitModel();
         $seen = [];
         $projected = [];
+        $existing = $waitModel::query()->where('workflow_run_id', $run->id)->get()->keyBy('id');
 
         foreach (array_values($waits) as $position => $wait) {
             $waitId = self::waitId($wait, $position);
@@ -186,6 +187,7 @@ public static function project(WorkflowRun $run, ?array $waits = null): array
                     'history_unsupported_reason' => self::stringValue($wait['history_unsupported_reason'] ?? null),
                     'payload' => $payload,
                 ],
+                $existing->get($projectionId),
             );
 
             $projected[] = $row;

src/V2/Support/RunWaitView.php

@@ -17,9 +17,11 @@
 final class RunWaitView
 {
     /**
+     * @param list<array<string, mixed>>|null $activities Metadata already derived for this projection pass.
+     * @param list<array<string, mixed>>|null $timers
      * @return list<array<string, mixed>>
      */
-    public static function forRun(WorkflowRun $run): array
+    public static function forRun(WorkflowRun $run, ?array $activities = null, ?array $timers = null): array
     {
         $run->loadMissing([
             'historyEvents',
@@ -51,7 +53,7 @@ public static function forRun(WorkflowRun $run): array
 
         $waits = [];
 
-        foreach (RunActivityView::activitiesForRun($run, decodePayloads: false) as $activity) {
+        foreach ($activities ?? RunActivityView::activitiesForRun($run, decodePayloads: false) as $activity) {
             if (! is_string($activity['id'] ?? null)) {
                 continue;
             }
@@ -62,7 +64,7 @@ public static function forRun(WorkflowRun $run): array
         $waits = array_merge($waits, self::conditionWaits($conditionWaits, $taskByTimerId));
         $waits = array_merge($waits, UpdateWaits::forRun($run));
 
-        foreach (RunTimerView::timersForRun($run) as $timer) {
+        foreach ($timers ?? RunTimerView::timersForRun($run) as $timer) {
             if (
                 in_array($timer['id'] ?? null, $conditionTimerIds, true)
                 || ($timer['timer_kind'] ?? null) === 'condition_timeout'

tests/Unit/V2/IdempotentProjectionUpsertTest.php

@@ -15,6 +15,88 @@
 
 final class IdempotentProjectionUpsertTest extends TestCase
 {
+    public function testPrefetchedRowRetainsModelHooksWithoutAnotherSelect(): void
+    {
+        $run = $this->seedRun();
+        $id = hash('sha256', $run->id . '|prefetched');
+        $row = IdempotentProjectionUpsert::upsert(
+            WorkflowTimelineEntry::class,
+            [
+                'id' => $id,
+            ],
+            $this->timelineAttributes($run, 'prefetched', 'before'),
+        );
+        $calls = 0;
+        WorkflowTimelineEntry::saving(static function (WorkflowTimelineEntry $entry) use ($id, &$calls): void {
+            if ($entry->id === $id) {
+                $calls++;
+            }
+        });
+        $connection = $row->getConnection();
+        $connection->flushQueryLog();
+        $connection->enableQueryLog();
+
+        try {
+            $updated = IdempotentProjectionUpsert::upsert(
+                WorkflowTimelineEntry::class,
+                [
+                    'id' => $id,
+                ],
+                [
+                    'summary' => 'after',
+                ],
+                $row,
+            );
+            $queries = $connection->getQueryLog();
+        } finally {
+            $connection->disableQueryLog();
+            WorkflowTimelineEntry::flushEventListeners();
+        }
+
+        $this->assertSame($row, $updated);
+        $this->assertSame(1, $calls);
+        $this->assertSame('after', $row->fresh()->summary);
+        $this->assertCount(0, array_filter($queries, static fn (array $query): bool =>
+            str_starts_with(strtolower($query['query']), 'select')));
+    }
+
+    public function testRejectsPrefetchedRowWithDifferentIdentity(): void
+    {
+        $run = $this->seedRun();
+        $row = IdempotentProjectionUpsert::upsert(
+            WorkflowTimelineEntry::class,
+            [
+                'id' => hash('sha256', 'original'),
+            ],
+            $this->timelineAttributes($run, 'prefetched', 'original'),
+        );
+
+        $this->expectException(\InvalidArgumentException::class);
+        IdempotentProjectionUpsert::upsert(
+            WorkflowTimelineEntry::class,
+            [
+                'id' => hash('sha256', 'different'),
+            ],
+            [
+                'summary' => 'must not write',
+            ],
+            $row,
+        );
+    }
+
+    public function testRejectsUnpersistedPrefetchedRow(): void
+    {
+        $this->expectException(\InvalidArgumentException::class);
+        IdempotentProjectionUpsert::upsert(
+            WorkflowTimelineEntry::class,
+            [
+                'id' => hash('sha256', 'unpersisted'),
+            ],
+            [],
+            new WorkflowTimelineEntry(),
+        );
+    }
+
     public function testInsertsRowWhenNoConflictExists(): void
     {
         $run = $this->seedRun();