From 9b841a3f2deaad7150cd462f9f094b6bc03255f7 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 11 Sep 2026 18:41:15 +0000 Subject: [PATCH 1/2] Reuse existing rows and derived views within projection passes --- src/V2/Support/IdempotentProjectionUpsert.php | 24 +- src/V2/Support/RunSummaryProjector.php | 4 +- src/V2/Support/RunTimelineProjector.php | 12 +- src/V2/Support/RunTimerProjector.php | 2 + src/V2/Support/RunWaitProjector.php | 2 + src/V2/Support/RunWaitView.php | 8 +- .../V2/IdempotentProjectionUpsertTest.php | 82 +++++++ tests/Unit/V2/ProjectionPrefetchTest.php | 214 ++++++++++++++++++ tests/Unit/V2/ProjectionReplayTest.php | 96 ++++++++ 9 files changed, 437 insertions(+), 7 deletions(-) create mode 100644 tests/Unit/V2/ProjectionPrefetchTest.php create mode 100644 tests/Unit/V2/ProjectionReplayTest.php diff --git a/src/V2/Support/IdempotentProjectionUpsert.php b/src/V2/Support/IdempotentProjectionUpsert.php index 0932f80a..df39a702 100644 --- a/src/V2/Support/IdempotentProjectionUpsert.php +++ b/src/V2/Support/IdempotentProjectionUpsert.php @@ -30,11 +30,33 @@ final class IdempotentProjectionUpsert * @param class-string $model * @param array $key * @param array $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); diff --git a/src/V2/Support/RunSummaryProjector.php b/src/V2/Support/RunSummaryProjector.php index 26dd7161..a16c4685 100644 --- a/src/V2/Support/RunSummaryProjector.php +++ b/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; diff --git a/src/V2/Support/RunTimelineProjector.php b/src/V2/Support/RunTimelineProjector.php index aa315d6c..e8abeb15 100644 --- a/src/V2/Support/RunTimelineProjector.php +++ b/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; diff --git a/src/V2/Support/RunTimerProjector.php b/src/V2/Support/RunTimerProjector.php index 6928a1db..95ffe882 100644 --- a/src/V2/Support/RunTimerProjector.php +++ b/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; diff --git a/src/V2/Support/RunWaitProjector.php b/src/V2/Support/RunWaitProjector.php index 98bf9cdb..2184a26c 100644 --- a/src/V2/Support/RunWaitProjector.php +++ b/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; diff --git a/src/V2/Support/RunWaitView.php b/src/V2/Support/RunWaitView.php index 415793ac..c161b3e4 100644 --- a/src/V2/Support/RunWaitView.php +++ b/src/V2/Support/RunWaitView.php @@ -17,9 +17,11 @@ final class RunWaitView { /** + * @param list>|null $activities Metadata already derived for this projection pass. + * @param list>|null $timers * @return list> */ - 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' diff --git a/tests/Unit/V2/IdempotentProjectionUpsertTest.php b/tests/Unit/V2/IdempotentProjectionUpsertTest.php index b3469bbf..1574c359 100644 --- a/tests/Unit/V2/IdempotentProjectionUpsertTest.php +++ b/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(); diff --git a/tests/Unit/V2/ProjectionPrefetchTest.php b/tests/Unit/V2/ProjectionPrefetchTest.php new file mode 100644 index 00000000..60863b56 --- /dev/null +++ b/tests/Unit/V2/ProjectionPrefetchTest.php @@ -0,0 +1,214 @@ + + */ + public static function projectors(): iterable + { + yield 'timeline' => [RunTimelineProjector::class, 'history_event_id']; + yield 'wait' => [RunWaitProjector::class, 'wait_id']; + yield 'timer' => [RunTimerProjector::class, 'timer_id']; + } + + /** + * @return iterable}> + */ + public static function configuredProjectors(): iterable + { + yield 'timeline' => [RunTimelineProjector::class, 'run_timeline_entry_model', PrefetchedTimelineEntry::class]; + yield 'wait' => [RunWaitProjector::class, 'run_wait_model', PrefetchedRunWait::class]; + yield 'timer' => [RunTimerProjector::class, 'run_timer_entry_model', PrefetchedTimerEntry::class]; + } + + #[DataProvider('configuredProjectors')] + public function testPrefetchUsesConfiguredModelTableAndConnection( + string $projector, + string $configKey, + string $model + ): void { + $run = $this->seedRun('configured'); + $entries = $this->entries(); + $original = $projector::project($run, $entries)[0]; + config() + ->set('database.connections.projection-secondary', [ + 'driver' => 'sqlite', + 'database' => ':memory:', + ]); + Schema::connection('projection-secondary')->create('custom_projection_rows', static function (Blueprint $table) use ( + $original + ): void { + $table->string('id') + ->primary(); + + foreach (array_keys($original->getAttributes()) as $column) { + if ($column !== 'id') { + $table->text($column) + ->nullable(); + } + } + }); + config() + ->set('workflows.v2.' . $configKey, $model); + + try { + $created = $projector::project($run->fresh(), $entries)[0]; + $entries[0]['status'] = 'updated'; + $updated = $projector::project($run->fresh(), $entries)[0]; + + $this->assertInstanceOf($model, $updated); + $this->assertSame($created->getKey(), $updated->getKey()); + $this->assertSame('updated', $updated->fresh()->payload['status']); + $this->assertSame('resolved', $original->fresh()->payload['status']); + $this->assertSame( + count($entries), + DB::connection('projection-secondary')->table('custom_projection_rows')->count() + ); + $projector::project($run->fresh(), []); + $this->assertSame(0, DB::connection('projection-secondary')->table('custom_projection_rows')->count()); + $this->assertNotNull($original->fresh()); + } finally { + DB::purge('projection-secondary'); + } + } + + #[DataProvider('projectors')] + public function testReprojectionUsesOneScopedReadAndPreservesRows(string $projector, string $identity): void + { + $run = $this->seedRun('prefetch'); + $entries = $this->entries(); + /** @var list $rows */ + $rows = $projector::project($run, $entries); + $expected = array_map(self::attributes(...), $rows); + $connection = $rows[0]->getConnection(); + $table = $rows[0]->getTable(); + $connection->flushQueryLog(); + $connection->enableQueryLog(); + + try { + $reprojected = $projector::project($run->fresh(), $entries); + $queries = $connection->getQueryLog(); + } finally { + $connection->disableQueryLog(); + } + + $reads = array_filter($queries, static fn (array $query): bool => + str_starts_with(strtolower($query['query']), 'select') && str_contains($query['query'], $table)); + // One prefetched row set plus the existing stale-cleanup primary-key snapshot. + $this->assertCount(2, $reads); + $this->assertSame($expected, array_map(self::attributes(...), $reprojected)); + + $rows[0]->forceFill([ + 'payload' => [ + 'corrupt' => true, + ], + ])->save(); + $rows[1]->delete(); + $orphan = $rows[2]->replicate(); + $orphan->forceFill([ + 'id' => hash('sha256', 'orphan'), + $identity => 'orphan', + ])->save(); + $otherRun = $this->seedRun('unrelated'); + $otherRows = $projector::project($otherRun, $entries); + + $repaired = $projector::project($run->fresh(), $entries); + $this->assertCount(count($entries), $repaired); + $this->assertSame($expected[0]['payload'], $repaired[0]->getRawOriginal('payload')); + $this->assertSame($expected[1]['id'], $repaired[1]->getKey()); + $this->assertNull($orphan->fresh()); + $this->assertNotNull($otherRows[0]->fresh()); + + $this->assertSame([], $projector::project($run->fresh(), [])); + $this->assertSame(0, $rows[0]->newQuery()->where('workflow_run_id', $run->id)->count()); + $this->assertSame(count($entries), $otherRows[0]->newQuery()->where('workflow_run_id', $otherRun->id)->count()); + } + + /** + * @return list> + */ + private function entries(): array + { + return array_map(static fn (int $sequence): array => [ + 'id' => 'entry-' . $sequence, + 'sequence' => $sequence, + 'type' => 'TimerFired', + 'kind' => 'timer', + 'status' => 'resolved', + 'summary' => 'Timer completed.', + 'recorded_at' => '2026-09-01T12:00:00.123456Z', + 'opened_at' => '2026-09-01T12:00:00.123456Z', + 'fire_at' => '2026-09-01T12:00:01.123456Z', + ], range(1, 20)); + } + + /** + * @return array + */ + private static function attributes(Model $row): array + { + $attributes = $row->getAttributes(); + ksort($attributes); + + return $attributes; + } + + private function seedRun(string $id): WorkflowRun + { + $instance = WorkflowInstance::query()->create([ + 'id' => $id, + 'workflow_class' => 'ProjectionWorkflow', + 'workflow_type' => 'projection.workflow', + 'namespace' => 'default', + ]); + + return WorkflowRun::query()->create([ + 'workflow_instance_id' => $instance->id, + 'run_number' => 1, + 'workflow_class' => $instance->workflow_class, + 'workflow_type' => $instance->workflow_type, + 'status' => 'waiting', + ]); + } +} + +final class PrefetchedTimelineEntry extends WorkflowTimelineEntry +{ + protected $table = 'custom_projection_rows'; + + protected $connection = 'projection-secondary'; +} + +final class PrefetchedRunWait extends WorkflowRunWait +{ + protected $table = 'custom_projection_rows'; + + protected $connection = 'projection-secondary'; +} + +final class PrefetchedTimerEntry extends WorkflowRunTimerEntry +{ + protected $table = 'custom_projection_rows'; + + protected $connection = 'projection-secondary'; +} diff --git a/tests/Unit/V2/ProjectionReplayTest.php b/tests/Unit/V2/ProjectionReplayTest.php new file mode 100644 index 00000000..17a738f6 --- /dev/null +++ b/tests/Unit/V2/ProjectionReplayTest.php @@ -0,0 +1,96 @@ +freezeTime(); + $workflow = WorkflowStub::make(ProjectionReplayWorkflow::class, 'projection-replay'); + $workflow->start($rounds); + + for ($round = 0; $round <= $rounds; $round++) { + WorkflowStub::runReadyTasks(); + $run = WorkflowRun::query()->findOrFail($workflow->runId()); + $activities = RunActivityView::activitiesForRun($run, decodePayloads: false); + $timers = RunTimerView::timersForRun($run); + + $this->assertEquals( + RunWaitView::forRun($run->fresh()), + RunWaitView::forRun($run, $activities, $timers), + ); + RunSummaryProjector::project($run->fresh()); + $this->assertFalse(RunTimelineProjector::driftStatusForRun($run->fresh())['stale']); + $this->assertFalse(RunWaitProjector::driftStatusForRun($run->fresh())['stale']); + $this->assertFalse(RunTimerProjector::driftStatusForRun($run->fresh())['stale']); + $this->travel(2) + ->seconds(); + } + + $workflow = WorkflowStub::load('projection-replay'); + $run = WorkflowRun::query()->findOrFail($workflow->runId()); + $this->assertTrue($workflow->completed()); + $this->assertSame($rounds * 3, $calls); + $this->assertSame($rounds * 3, $workflow->output()); + $this->assertSame( + $rounds * 3, + $run->historyEvents()->where('event_type', HistoryEventType::ActivityCompleted)->count() + ); + $this->assertSame($rounds, $run->historyEvents()->where('event_type', HistoryEventType::TimerFired)->count()); + } +} + +final class ProjectionReplayWorkflow extends Workflow +{ + public function handle(int $rounds): int + { + $result = 0; + + for ($round = 0; $round < $rounds; $round++) { + $result += array_sum(all([ + static fn (): mixed => activity(ProjectionReplayActivity::class, 1), + static fn (): mixed => activity(ProjectionReplayActivity::class, 1), + static fn (): mixed => activity(ProjectionReplayActivity::class, 1), + ])); + timer(1); + } + + return $result; + } +} + +final class ProjectionReplayActivity extends Activity +{ + public function handle(int $value): int + { + return $value; + } +} From c59828f4c7be77c6efd078419d0aed548c26c8fe Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 11 Sep 2026 18:46:25 +0000 Subject: [PATCH 2/2] Align replay regression assertion formatting --- tests/Unit/V2/ProjectionReplayTest.php | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/Unit/V2/ProjectionReplayTest.php b/tests/Unit/V2/ProjectionReplayTest.php index 17a738f6..462b9bbb 100644 --- a/tests/Unit/V2/ProjectionReplayTest.php +++ b/tests/Unit/V2/ProjectionReplayTest.php @@ -62,7 +62,8 @@ public function testRepeatedActivityAndTimerRoundsKeepColdProjectionsConsistent( $this->assertSame($rounds * 3, $workflow->output()); $this->assertSame( $rounds * 3, - $run->historyEvents()->where('event_type', HistoryEventType::ActivityCompleted)->count() + $run->historyEvents() + ->where('event_type', HistoryEventType::ActivityCompleted)->count() ); $this->assertSame($rounds, $run->historyEvents()->where('event_type', HistoryEventType::TimerFired)->count()); }