From bb7187000363b7bc8b1a9b13ddef4d703a2fba58 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 26 Oct 2026 13:02:32 +0000 Subject: [PATCH 1/2] Avoid rewriting unchanged retained wait projections --- src/V2/Models/WorkflowRun.php | 3 +- src/V2/Models/WorkflowRunWait.php | 61 ++++ src/V2/Support/RunWaitProjector.php | 35 ++- tests/Unit/V2/RunWaitProjectionWritesTest.php | 285 ++++++++++++++++++ 4 files changed, 379 insertions(+), 5 deletions(-) create mode 100644 tests/Unit/V2/RunWaitProjectionWritesTest.php diff --git a/src/V2/Models/WorkflowRun.php b/src/V2/Models/WorkflowRun.php index 7862e4a56..bf7e61f3e 100644 --- a/src/V2/Models/WorkflowRun.php +++ b/src/V2/Models/WorkflowRun.php @@ -217,8 +217,7 @@ public function waits(): HasMany ConfiguredV2Models::resolve('run_wait_model', WorkflowRunWait::class), 'workflow_run_id', ) - ->orderBy('position') - ->orderBy('wait_id'); + ->inDisplayOrder(); } public function timelineEntries(): HasMany diff --git a/src/V2/Models/WorkflowRunWait.php b/src/V2/Models/WorkflowRunWait.php index 8105456b5..3492f2e22 100644 --- a/src/V2/Models/WorkflowRunWait.php +++ b/src/V2/Models/WorkflowRunWait.php @@ -5,6 +5,7 @@ namespace Workflow\V2\Models; use Carbon\CarbonInterface; +use Illuminate\Database\Eloquent\Builder; use Illuminate\Database\Eloquent\Model; use Illuminate\Database\Eloquent\Relations\BelongsTo; use Illuminate\Support\Carbon; @@ -38,6 +39,53 @@ class WorkflowRunWait extends Model 'resolved_at' => UtcScheduleTimestamp::class, ]; + /** + * @param string $key + */ + public function originalIsEquivalent($key): bool + { + if (parent::originalIsEquivalent($key)) { + return true; + } + + if (! array_key_exists($key, $this->original)) { + return false; + } + + if ($key === 'payload' && ($this->getCasts()[$key] ?? null) === 'array') { + // MySQL normalizes JSON object key order. Lists and scalar types + // still carry meaning, so compare those without coercion. + return self::canonicalPayload($this->getAttribute($key)) + === self::canonicalPayload($this->getOriginal($key)); + } + + if (in_array($key, ['opened_at', 'deadline_at', 'resolved_at'], true) + && ($this->getCasts()[$key] ?? null) === UtcScheduleTimestamp::class) { + $current = $this->getAttribute($key); + $original = $this->getOriginal($key); + + // PostgreSQL can return a zero-fraction timestamp without its + // trailing digits. Its instant has not changed. + return $current instanceof CarbonInterface + && $original instanceof CarbonInterface + && $current->equalTo($original); + } + + return false; + } + + public function scopeInDisplayOrder(Builder $query): Builder + { + $status = $query->getQuery() + ->getGrammar() + ->wrap($this->qualifyColumn('status')); + + return $query->orderByRaw( + "CASE {$status} WHEN 'open' THEN 0 WHEN 'resolved' THEN 1 WHEN 'cancelled' THEN 2 ELSE 99 END", + )->orderBy($this->qualifyColumn('position')) + ->orderBy($this->qualifyColumn('wait_id')); + } + public function run(): BelongsTo { return $this->belongsTo( @@ -93,4 +141,17 @@ private static function timestamp(mixed $value): ?CarbonInterface ? Carbon::parse($value) : null; } + + private static function canonicalPayload(mixed $value): mixed + { + if (! is_array($value)) { + return $value; + } + + if (! array_is_list($value)) { + ksort($value); + } + + return array_map(self::canonicalPayload(...), $value); + } } diff --git a/src/V2/Support/RunWaitProjector.php b/src/V2/Support/RunWaitProjector.php index 345cea677..8fb9ac961 100644 --- a/src/V2/Support/RunWaitProjector.php +++ b/src/V2/Support/RunWaitProjector.php @@ -144,6 +144,7 @@ public static function project(WorkflowRun $run, ?array $waits = null): array $seen = []; $projected = []; $existing = $waitModel::query()->where('workflow_run_id', $run->id)->get()->keyBy('id'); + $positions = self::chronologicalPositions($waits); foreach (array_values($waits) as $position => $wait) { $waitId = self::waitId($wait, $position); @@ -161,7 +162,7 @@ public static function project(WorkflowRun $run, ?array $waits = null): array 'workflow_run_id' => $run->id, 'workflow_instance_id' => $run->workflow_instance_id, 'wait_id' => $waitId, - 'position' => $position, + 'position' => $positions[$waitId], 'kind' => self::stringValue($wait['kind'] ?? null) ?? 'unknown', 'sequence' => self::intValue($wait['sequence'] ?? null), 'status' => self::stringValue($wait['status'] ?? null) ?? 'unknown', @@ -296,13 +297,41 @@ private static function projectedRows(WorkflowRun $run): EloquentCollection /** @var EloquentCollection $waits */ $waits = $waitModel::query() ->where('workflow_run_id', $run->id) - ->orderBy('position') - ->orderBy('wait_id') + ->inDisplayOrder() ->get(); return $waits; } + /** + * Keep stored order independent of status. Otherwise closing one open + * wait and opening the next rewrites every retained row twice. + * + * @param list> $waits + * @return array + */ + private static function chronologicalPositions(array $waits): array + { + $ordered = []; + + foreach (array_values($waits) as $position => $wait) { + $ordered[] = [ + 'id' => self::waitId($wait, $position), + 'opened_at' => self::timestamp($wait['opened_at'] ?? null)?->getTimestampMs() ?? PHP_INT_MAX, + ]; + } + + usort($ordered, static fn (array $left, array $right): int => + ($left['opened_at'] <=> $right['opened_at']) ?: ($left['id'] <=> $right['id'])); + $positions = []; + + foreach ($ordered as $position => $wait) { + $positions[$wait['id']] = $position; + } + + return $positions; + } + /** * @param EloquentCollection $projected * @param list> $canonical diff --git a/tests/Unit/V2/RunWaitProjectionWritesTest.php b/tests/Unit/V2/RunWaitProjectionWritesTest.php new file mode 100644 index 000000000..f844e78db --- /dev/null +++ b/tests/Unit/V2/RunWaitProjectionWritesTest.php @@ -0,0 +1,285 @@ +setRawAttributes([ + 'payload' => '{"kind":"signal","extra":{"a":1,"b":[1,2]},"id":"signal:1"}', + ], true); + $row->payload = [ + 'id' => 'signal:1', + 'extra' => [ + 'b' => [1, 2], + 'a' => 1, + ], + 'kind' => 'signal', + ]; + + $this->assertFalse($row->isDirty('payload')); + + $row->payload = [ + 'id' => 'signal:1', + 'extra' => [ + 'b' => [2, 1], + 'a' => 1, + ], + 'kind' => 'signal', + ]; + $this->assertTrue($row->isDirty('payload')); + $row->payload = [ + 'id' => 'signal:1', + 'extra' => [ + 'b' => [1, 2], + 'a' => '1', + ], + 'kind' => 'signal', + ]; + $this->assertTrue($row->isDirty('payload')); + $row->payload = [ + 'id' => 'signal:1', + 'extra' => [ + 'b' => [1, 2], + 'a' => 1, + 'value' => null, + ], + 'kind' => 'signal', + ]; + $this->assertTrue($row->isDirty('payload')); + } + + public function testTimestampPrecisionDoesNotDirtyTheSameInstant(): void + { + $row = new WorkflowRunWait(); + $row->setRawAttributes([ + 'opened_at' => '2026-10-26 12:00:00', + 'deadline_at' => '2026-10-26 12:00:01.123456', + 'resolved_at' => null, + ], true); + $row->opened_at = Carbon::parse('2026-10-26T14:00:00+02:00'); + $row->deadline_at = Carbon::parse('2026-10-26T12:00:01.123456Z'); + $row->resolved_at = null; + + $this->assertFalse($row->isDirty()); + + $row->deadline_at = Carbon::parse('2026-10-26T12:00:01.123457Z'); + $this->assertTrue($row->isDirty('deadline_at')); + $row->resolved_at = Carbon::parse('2026-10-26T12:00:02Z'); + $this->assertTrue($row->isDirty('resolved_at')); + $row->opened_at = null; + $this->assertTrue($row->isDirty('opened_at')); + } + + public function testUnchangedRetainedRowsIssueNoUpdatesAfterStorageReordersJson(): void + { + $this->travelTo(Carbon::parse('2026-10-26T12:00:00Z')); + $run = $this->runFixture(); + $waits = []; + for ($index = 0; $index < 64; $index++) { + $waits[] = $this->wait('signal:' . $index, $index, 'resolved'); + } + RunWaitProjector::project($run, $waits); + foreach (WorkflowRunWait::query()->where('workflow_run_id', $run->id)->get() as $row) { + $row->getConnection() + ->table($row->getTable()) + ->where('id', $row->id) + ->update([ + 'payload' => json_encode(array_reverse($row->payload, true), JSON_THROW_ON_ERROR), + ]); + } + $before = $this->rows($run); + $this->travel(10) + ->seconds(); + + $queries = $this->capture(static fn () => RunWaitProjector::project($run->fresh(), $waits)); + + $this->assertSame(0, $this->updateCount($queries)); + $this->assertSame($before, $this->rows($run)); + + $waits[32]['summary'] = 'A genuine changed wait'; + $queries = $this->capture(static fn () => RunWaitProjector::project($run->fresh(), $waits)); + $this->assertSame(1, $this->updateCount($queries)); + $this->assertSame( + 'A genuine changed wait', + $this->rows($run)[hash('sha256', $run->id . '|signal:32')]['summary'] + ); + } + + public function testStatusChangesPreserveDisplayOrderWithoutRepositioningRetainedRows(): void + { + $this->travelTo(Carbon::parse('2026-10-26T12:00:00Z')); + $run = $this->runFixture(); + $old = $this->wait('signal:old', 0, 'resolved'); + $current = $this->wait('signal:current', 1, 'open'); + $cancelled = $this->wait('signal:cancelled', 2, 'cancelled'); + RunWaitProjector::project($run, [$current, $old, $cancelled]); + $before = $this->rows($run); + $this->assertSame( + ['signal:current', 'signal:old', 'signal:cancelled'], + $run->fresh()->waits->pluck('wait_id')->all() + ); + $this->travel(10) + ->seconds(); + $current['status'] = 'resolved'; + $current['source_status'] = 'received'; + $current['resolved_at'] = now(); + $next = $this->wait('signal:next', 3, 'open'); + + $queries = $this->capture( + static fn () => RunWaitProjector::project($run->fresh(), [$next, $old, $current, $cancelled]) + ); + + $this->assertSame(1, $this->updateCount($queries)); + $after = $this->rows($run); + foreach (['signal:old', 'signal:cancelled'] as $id) { + $key = hash('sha256', $run->id . '|' . $id); + $this->assertSame($before[$key], $after[$key]); + } + $this->assertSame( + ['signal:next', 'signal:old', 'signal:current', 'signal:cancelled'], + $run->fresh() + ->waits->pluck('wait_id') + ->all(), + ); + } + + public function testProjectionStillRunsModelHooksAndPersistsTheirChanges(): void + { + $run = $this->runFixture(); + $waits = [$this->wait('signal:hook', 1, 'resolved')]; + RunWaitProjector::project($run, $waits); + $calls = 0; + WorkflowRunWait::saving(static function (WorkflowRunWait $row) use (&$calls): void { + $calls++; + $row->summary = 'Model hook changed this field'; + }); + + try { + $queries = $this->capture(static fn () => RunWaitProjector::project($run->fresh(), $waits)); + } finally { + WorkflowRunWait::flushEventListeners(); + } + + $this->assertSame(1, $calls); + $this->assertSame(1, $this->updateCount($queries)); + $this->assertSame('Model hook changed this field', $run->fresh()->waits->first()->summary); + } + + public function testMissingRowsAreRecreatedAndRealChangesRollBack(): void + { + $run = $this->runFixture(); + $waits = [$this->wait('signal:one', 0, 'resolved'), $this->wait('signal:two', 1, 'open')]; + RunWaitProjector::project($run, $waits); + $missing = hash('sha256', $run->id . '|signal:one'); + WorkflowRunWait::query()->whereKey($missing)->delete(); + RunWaitProjector::project($run->fresh(), $waits); + $this->assertSame(2, $run->fresh()->waits()->count()); + $this->assertNotNull(WorkflowRunWait::query()->find($missing)); + $before = $this->rows($run); + $connection = $run->getConnection(); + $connection->beginTransaction(); + + try { + $waits[1]['summary'] = 'Rolled back transition'; + RunWaitProjector::project($run->fresh(), $waits); + } finally { + $connection->rollBack(); + } + + $this->assertSame($before, $this->rows($run)); + } + + private function runFixture(): WorkflowRun + { + $instance = WorkflowInstance::query()->create([ + 'id' => (string) Str::ulid(), + 'workflow_class' => 'App\\WaitWriteFixture', + 'workflow_type' => 'App\\WaitWriteFixture', + 'namespace' => 'default', + ]); + + return WorkflowRun::query()->create([ + 'id' => (string) Str::ulid(), + 'workflow_instance_id' => $instance->id, + 'run_number' => 1, + 'workflow_class' => $instance->workflow_class, + 'workflow_type' => $instance->workflow_type, + 'status' => 'running', + ]); + } + + /** + * @return array + */ + private function wait(string $id, int $sequence, string $status): array + { + return [ + 'id' => $id, + 'kind' => 'signal', + 'sequence' => $sequence, + 'status' => $status, + 'source_status' => $status === 'open' ? 'waiting' : 'received', + 'summary' => 'Signal wait ' . $id, + 'opened_at' => Carbon::parse('2026-10-26T10:00:00Z')->addSeconds($sequence), + 'resolved_at' => $status === 'open' ? null : Carbon::parse('2026-10-26T11:00:00Z')->addSeconds($sequence), + 'task_backed' => false, + 'external_only' => true, + ]; + } + + /** + * @return array> + */ + private function rows(WorkflowRun $run): array + { + return $run->getConnection() + ->table('workflow_run_waits') + ->where('workflow_run_id', $run->id) + ->orderBy('id') + ->get() + ->mapWithKeys(static fn ($row): array => [ + $row->id => (array) $row, + ])->all(); + } + + /** + * @return list> + */ + private function capture(callable $operation): array + { + $connection = DB::connection(); + $connection->flushQueryLog(); + $connection->enableQueryLog(); + try { + $operation(); + + return $connection->getQueryLog(); + } finally { + $connection->disableQueryLog(); + } + } + + /** + * @param list> $queries + */ + private function updateCount(array $queries): int + { + return count(array_filter($queries, static fn (array $query): bool => + str_starts_with(strtolower(ltrim($query['query'])), 'update ') + && str_contains($query['query'], 'workflow_run_waits'))); + } +} From c608431d4250b731b6f261fca540d59028006712 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 26 Oct 2026 13:20:50 +0000 Subject: [PATCH 2/2] Qualify retained wait writes across storage and workflow progress --- .../V2/TestRetainedSignalWritesWorkflow.php | 24 ++ tests/Unit/V2/RunWaitProjectionWritesTest.php | 250 +++++++++++++++++- 2 files changed, 271 insertions(+), 3 deletions(-) create mode 100644 tests/Fixtures/V2/TestRetainedSignalWritesWorkflow.php diff --git a/tests/Fixtures/V2/TestRetainedSignalWritesWorkflow.php b/tests/Fixtures/V2/TestRetainedSignalWritesWorkflow.php new file mode 100644 index 000000000..376022002 --- /dev/null +++ b/tests/Fixtures/V2/TestRetainedSignalWritesWorkflow.php @@ -0,0 +1,24 @@ +set('workflows.v2.task_dispatch_mode', 'poll'); + Queue::fake(); + $this->travelTo(Carbon::parse('2026-10-26T12:00:00Z')); + $workflow = WorkflowStub::make(TestRetainedSignalWritesWorkflow::class); + $workflow->start(17); + $run = $workflow->run(); + $this->assertInstanceOf(WorkflowRun::class, $run); + $this->drainTasks($run); + + for ($step = 1; $step <= 17; $step++) { + $before = $this->rows($run); + $this->travel(10) + ->seconds(); + $queries = $this->capture(function () use ($workflow, $step, $run): void { + $workflow->signal('advance', $step); + $this->drainTasks($run); + }); + $after = $this->rows($run); + $this->assertCount(min(17, $step + 1), $after); + + foreach ($before as $id => $row) { + if ($row['status'] === 'resolved') { + $this->assertSame($row, $after[$id], 'Retained wait changed at signal ' . $step); + } + } + + if (in_array($step, [1, 8, 16, 17], true)) { + // Signal acceptance and task completion update the selected + // wait twice. The retained count must not increase the writes. + $this->assertSame(2, $this->updateCount($queries)); + $snapshot = RunWaitProjector::snapshotForRun($run->fresh()); + $this->assertSame('workflow_run_waits', $snapshot['source']); + $this->assertSame( + array_column(RunWaitView::forRun($run->fresh()), 'id'), + array_column($snapshot['waits'], 'id'), + ); + } + } + + $this->assertTrue($workflow->refresh()->completed()); + $this->assertSame(range(1, 17), $workflow->output()); + } + public function testJsonObjectKeyOrderDoesNotDirtyAnUnchangedWait(): void { $row = new WorkflowRunWait(); @@ -84,6 +136,30 @@ public function testTimestampPrecisionDoesNotDirtyTheSameInstant(): void $this->assertTrue($row->isDirty('opened_at')); } + public function testEquivalenceKeepsNullsAndConfiguredCastSemantics(): void + { + $row = new WorkflowRunWait(); + $row->setRawAttributes([ + 'payload' => null, + ], true); + $row->payload = []; + $this->assertTrue($row->isDirty('payload')); + $row->payload = null; + $this->assertFalse($row->isDirty('payload')); + $row->mergeCasts([ + 'payload' => 'string', + 'opened_at' => 'string', + ]); + $row->setRawAttributes([ + 'payload' => '{"a":1,"b":2}', + 'opened_at' => '2026-10-26 12:00:00', + ], true); + $row->payload = '{"b":2,"a":1}'; + $row->opened_at = '2026-10-26 12:00:00.000000'; + $this->assertTrue($row->isDirty('payload')); + $this->assertTrue($row->isDirty('opened_at')); + } + public function testUnchangedRetainedRowsIssueNoUpdatesAfterStorageReordersJson(): void { $this->travelTo(Carbon::parse('2026-10-26T12:00:00Z')); @@ -130,7 +206,9 @@ public function testStatusChangesPreserveDisplayOrderWithoutRepositioningRetaine $before = $this->rows($run); $this->assertSame( ['signal:current', 'signal:old', 'signal:cancelled'], - $run->fresh()->waits->pluck('wait_id')->all() + $run->fresh() + ->waits->pluck('wait_id') + ->all() ); $this->travel(10) ->seconds(); @@ -179,6 +257,110 @@ public function testProjectionStillRunsModelHooksAndPersistsTheirChanges(): void $this->assertSame('Model hook changed this field', $run->fresh()->waits->first()->summary); } + public function testLegacyPositionsRemainReadableBeforeAndAfterReprojection(): void + { + $run = $this->runFixture(); + $resolved = $this->wait('signal:resolved', 0, 'resolved'); + $open = $this->wait('signal:open', 1, 'open'); + $cancelled = $this->wait('signal:cancelled', 2, 'cancelled'); + $waits = [$open, $resolved, $cancelled]; + RunWaitProjector::project($run, $waits); + + // Published versions store the display rank, including status priority. + foreach ($waits as $position => $wait) { + WorkflowRunWait::query()->where('workflow_run_id', $run->id) + ->where('wait_id', $wait['id']) + ->update([ + 'position' => $position, + ]); + } + $before = $this->rows($run); + $queries = $this->capture(fn () => $this->assertSame( + array_column($waits, 'id'), + $run->fresh() + ->waits->pluck('wait_id') + ->all(), + )); + $this->assertSame(0, $this->updateCount($queries)); + $this->assertSame($before, $this->rows($run)); + + RunWaitProjector::project($run->fresh(), $waits); + $queries = $this->capture(static fn () => RunWaitProjector::project($run->fresh(), $waits)); + $this->assertSame(0, $this->updateCount($queries)); + $this->assertSame(array_column($waits, 'id'), $run->fresh()->waits->pluck('wait_id')->all()); + } + + public function testDisplayOrderingPreservesTimestampTiesAndMissingTimes(): void + { + $run = $this->runFixture(); + $a = $this->wait('signal:a', 0, 'resolved'); + $z = $this->wait('signal:z', 0, 'resolved'); + $a['opened_at'] = Carbon::parse('2026-10-26T10:00:00.000400Z'); + $z['opened_at'] = Carbon::parse('2026-10-26T10:00:00.000100Z'); + $this->assertSame($a['opened_at']->getTimestampMs(), $z['opened_at']->getTimestampMs()); + $missing = $this->wait('signal:missing', 1, 'resolved'); + $missing['opened_at'] = null; + $open = $this->wait('signal:open', 2, 'open'); + RunWaitProjector::project($run, [$open, $a, $z, $missing]); + + // The canonical view uses millisecond precision, then the wait id. + $this->assertSame( + ['signal:open', 'signal:a', 'signal:z', 'signal:missing'], + $run->fresh() + ->waits->pluck('wait_id') + ->all(), + ); + $this->assertSame( + ['signal:a', 'signal:z', 'signal:open', 'signal:missing'], + WorkflowRunWait::query()->where('workflow_run_id', $run->id) + ->orderBy('position') + ->pluck('wait_id') + ->all(), + ); + } + + public function testConcurrentInsertFromAnotherConnectionRetainsWaitIdentityAndChanges(): void + { + $run = $this->runFixture(); + $waits = [$this->wait('signal:raced', 1, 'open')]; + $id = hash('sha256', $run->id . '|signal:raced'); + config() + ->set('database.connections.wait-writer', $run->getConnection()->getConfig()); + $raced = false; + WorkflowRunWait::saving(static function (WorkflowRunWait $row) use ($id, &$raced): void { + if ($row->exists || $row->id !== $id || $raced) { + return; + } + + $raced = true; + // The candidate has already observed absence. An independent + // connection commits the competing insert before its save. + DB::connection('wait-writer')->table($row->getTable())->insert( + array_replace($row->getAttributes(), [ + 'summary' => 'Competing writer', + 'created_at' => now() + ->format('Y-m-d H:i:s.u'), + 'updated_at' => now() + ->format('Y-m-d H:i:s.u'), + ]), + ); + }); + + try { + $row = RunWaitProjector::project($run, $waits)[0]; + } finally { + WorkflowRunWait::flushEventListeners(); + DB::purge('wait-writer'); + } + + $this->assertTrue($raced); + $this->assertSame($id, $row->id); + $this->assertSame($waits[0]['summary'], $row->fresh()->summary); + $this->assertSame(1, WorkflowRunWait::query()->where('workflow_run_id', $run->id)->count()); + $queries = $this->capture(static fn () => RunWaitProjector::project($run->fresh(), $waits)); + $this->assertSame(0, $this->updateCount($queries)); + } + public function testMissingRowsAreRecreatedAndRealChangesRollBack(): void { $run = $this->runFixture(); @@ -203,6 +385,51 @@ public function testMissingRowsAreRecreatedAndRealChangesRollBack(): void $this->assertSame($before, $this->rows($run)); } + public function testNoopAndChangedWritesUseTheConfiguredStorageConnection(): void + { + $run = $this->runFixture(); + $storage = $run->getConnection(); + $default = DB::getDefaultConnection(); + $previousStorage = config('workflows.storage.connection'); + config() + ->set('database.connections.wait-application', [ + 'driver' => 'sqlite', + 'database' => ':memory:', + 'prefix' => '', + ]); + config() + ->set('workflows.storage.connection', $storage->getName()); + DB::setDefaultConnection('wait-application'); + + try { + $this->assertFalse(DB::connection()->getSchemaBuilder()->hasTable('workflow_run_waits')); + $waits = [$this->wait('signal:stored', 1, 'resolved')]; + RunWaitProjector::project($run, $waits); + $before = $this->rows($run); + $queries = $this->capture(static fn () => RunWaitProjector::project($run->fresh(), $waits), $storage); + $this->assertSame(0, $this->updateCount($queries)); + $this->assertSame($before, $this->rows($run)); + $storage->beginTransaction(); + try { + $waits[0]['summary'] = 'Changed on storage connection'; + $queries = $this->capture( + static fn () => RunWaitProjector::project($run->fresh(), $waits), + $storage, + ); + $this->assertSame(1, $this->updateCount($queries)); + $this->assertSame($waits[0]['summary'], $run->fresh()->waits->first()->summary); + } finally { + $storage->rollBack(); + } + $this->assertSame($before, $this->rows($run)); + } finally { + DB::setDefaultConnection($default); + config() + ->set('workflows.storage.connection', $previousStorage); + DB::purge('wait-application'); + } + } + private function runFixture(): WorkflowRun { $instance = WorkflowInstance::query()->create([ @@ -222,6 +449,23 @@ private function runFixture(): WorkflowRun ]); } + private function drainTasks(WorkflowRun $run): void + { + for ($attempt = 0; $attempt < 8; $attempt++) { + $task = WorkflowTask::query()->where('workflow_run_id', $run->id) + ->where('status', 'ready') + ->first(); + if ($task === null) { + return; + } + + app() + ->call([new RunWorkflowTask($task->id), 'handle']); + } + + $this->fail('Workflow tasks exceeded the fixture drain bound.'); + } + /** * @return array */ @@ -259,9 +503,9 @@ private function rows(WorkflowRun $run): array /** * @return list> */ - private function capture(callable $operation): array + private function capture(callable $operation, ?Connection $connection = null): array { - $connection = DB::connection(); + $connection ??= DB::connection(); $connection->flushQueryLog(); $connection->enableQueryLog(); try {