Skip to content

Commit 3ecfb40

Browse files
durable-workflow-opsdurable-workflow-ops
committed
Route execution-plane projections through the history role
1 parent fddfed6 commit 3ecfb40

7 files changed

Lines changed: 235 additions & 32 deletions

‎src/V2/Jobs/RunWorkflowTask.php‎

Lines changed: 20 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,12 @@
1111
use Illuminate\Queue\SerializesModels;
1212
use Illuminate\Support\Facades\DB;
1313
use Throwable;
14+
use Workflow\V2\Contracts\HistoryProjectionRole;
1415
use Workflow\V2\Enums\TaskStatus;
1516
use Workflow\V2\Enums\TaskType;
1617
use Workflow\V2\Models\WorkflowRun;
1718
use Workflow\V2\Models\WorkflowTask;
1819
use Workflow\V2\Support\ConfiguredV2Models;
19-
use Workflow\V2\Support\RunSummaryProjector;
2020
use Workflow\V2\Support\TaskBackendCapabilities;
2121
use Workflow\V2\Support\TaskCompatibility;
2222
use Workflow\V2\Support\TaskDispatcher;
@@ -71,7 +71,7 @@ public function handle(WorkflowExecutor $executor): void
7171
'lease_expires_at' => null,
7272
])->save();
7373

74-
RunSummaryProjector::project(
74+
$this->projectRun(
7575
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures'])
7676
);
7777

@@ -99,7 +99,7 @@ public function handle(WorkflowExecutor $executor): void
9999

100100
/** @var WorkflowRun $run */
101101
$run = ConfiguredV2Models::query('run_model', WorkflowRun::class)->findOrFail($task->workflow_run_id);
102-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
102+
$this->projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
103103
});
104104

105105
throw $throwable;
@@ -128,13 +128,13 @@ private function claimTask(): bool
128128
TaskCompatibility::sync($task, $run);
129129

130130
if (TaskBackendCapabilities::recordClaimFailureIfUnsupported($task) !== null) {
131-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
131+
$this->projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
132132

133133
return false;
134134
}
135135

136136
if (! TaskCompatibility::supported($task, $run)) {
137-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
137+
$this->projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
138138

139139
return false;
140140
}
@@ -150,9 +150,23 @@ private function claimTask(): bool
150150
'last_claim_error' => null,
151151
])->save();
152152

153-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
153+
$this->projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
154154

155155
return true;
156156
});
157157
}
158+
159+
private function projectRun(WorkflowRun $run): void
160+
{
161+
$this->historyProjectionRole()
162+
->projectRun($run);
163+
}
164+
165+
private function historyProjectionRole(): HistoryProjectionRole
166+
{
167+
/** @var HistoryProjectionRole $role */
168+
$role = app(HistoryProjectionRole::class);
169+
170+
return $role;
171+
}
158172
}

‎src/V2/Support/ActivityOutcomeRecorder.php‎

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
use Throwable;
99
use Workflow\Serializers\CodecRegistry;
1010
use Workflow\Serializers\Serializer;
11+
use Workflow\V2\Contracts\HistoryProjectionRole;
1112
use Workflow\V2\Enums\ActivityAttemptStatus;
1213
use Workflow\V2\Enums\ActivityStatus;
1314
use Workflow\V2\Enums\FailureCategory;
@@ -92,7 +93,7 @@ public static function record(
9293

9394
ActivityCancellation::record($run, $lockedExecution, $task);
9495

95-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
96+
self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
9697

9798
return self::ignored($reason);
9899
}
@@ -167,7 +168,7 @@ public static function record(
167168
'lease_expires_at' => null,
168169
])->save();
169170

170-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
171+
self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
171172

172173
return self::recorded(null);
173174
}
@@ -272,7 +273,7 @@ public static function record(
272273
'activity' => ActivitySnapshot::fromExecution($lockedExecution),
273274
], $parallelMetadata ?? []), $task);
274275

275-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
276+
self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
276277

277278
return self::recorded($retryTask);
278279
} else {
@@ -359,7 +360,7 @@ public static function record(
359360
$closedStatus,
360361
)
361362
) {
362-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
363+
self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
363364

364365
return self::recorded(null);
365366
}
@@ -379,7 +380,7 @@ public static function record(
379380
'compatibility' => $run->compatibility,
380381
]);
381382

382-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
383+
self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
383384

384385
return self::recorded($resumeTask);
385386
});
@@ -573,4 +574,17 @@ private static function preferredPayloadCodec(ActivityExecution $execution, ?str
573574

574575
return CodecRegistry::defaultCodec();
575576
}
577+
578+
private static function projectRun(WorkflowRun $run): void
579+
{
580+
self::historyProjectionRole()->projectRun($run);
581+
}
582+
583+
private static function historyProjectionRole(): HistoryProjectionRole
584+
{
585+
/** @var HistoryProjectionRole $role */
586+
$role = app(HistoryProjectionRole::class);
587+
588+
return $role;
589+
}
576590
}

‎src/V2/Support/ActivityTimeoutEnforcer.php‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66

77
use Illuminate\Support\Facades\DB;
88
use Throwable;
9+
use Workflow\V2\Contracts\HistoryProjectionRole;
910
use Workflow\V2\Enums\ActivityAttemptStatus;
1011
use Workflow\V2\Enums\ActivityStatus;
1112
use Workflow\V2\Enums\FailureCategory;
@@ -273,7 +274,7 @@ private static function scheduleRetry(
273274
'activity' => ActivitySnapshot::fromExecution($execution),
274275
]), $existingTask);
275276

276-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
277+
self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
277278

278279
return [
279280
'enforced' => true,
@@ -368,7 +369,7 @@ private static function recordTerminalTimeout(
368369
ActivityStatus::Failed,
369370
)
370371
) {
371-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
372+
self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
372373

373374
return [
374375
'enforced' => true,
@@ -390,7 +391,7 @@ private static function recordTerminalTimeout(
390391
'compatibility' => $run->compatibility,
391392
]);
392393

393-
RunSummaryProjector::project($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
394+
self::projectRun($run->fresh(['instance', 'tasks', 'activityExecutions', 'failures']));
394395

395396
return [
396397
'enforced' => true,
@@ -471,4 +472,17 @@ private static function skipped(string $reason): array
471472
'next_task' => null,
472473
];
473474
}
475+
476+
private static function projectRun(WorkflowRun $run): void
477+
{
478+
self::historyProjectionRole()->projectRun($run);
479+
}
480+
481+
private static function historyProjectionRole(): HistoryProjectionRole
482+
{
483+
/** @var HistoryProjectionRole $role */
484+
$role = app(HistoryProjectionRole::class);
485+
486+
return $role;
487+
}
474488
}

‎src/V2/Support/WorkflowExecutor.php‎

Lines changed: 33 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
use Workflow\Serializers\CodecRegistry;
1414
use Workflow\Serializers\Serializer;
1515
use Workflow\V2\CommandContext;
16+
use Workflow\V2\Contracts\HistoryProjectionRole;
1617
use Workflow\V2\Enums\ActivityStatus;
1718
use Workflow\V2\Enums\ChildCallStatus;
1819
use Workflow\V2\Enums\CommandOutcome;
@@ -1713,7 +1714,7 @@ private function scheduleChildWorkflow(
17131714
$this->markRunWaiting($run, $task);
17141715
}
17151716

1716-
RunSummaryProjector::project(
1717+
$this->projectRun(
17171718
$childRun->fresh([
17181719
'instance',
17191720
'tasks',
@@ -2240,7 +2241,7 @@ private function markRunWaiting(WorkflowRun $run, WorkflowTask $task): void
22402241
'lease_expires_at' => null,
22412242
])->save();
22422243

2243-
RunSummaryProjector::project(
2244+
$this->projectRun(
22442245
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
22452246
);
22462247
}
@@ -2496,7 +2497,7 @@ private function continueAsNew(
24962497

24972498
LifecycleEventDispatcher::workflowStarted($continuedRun);
24982499

2499-
RunSummaryProjector::project(
2500+
$this->projectRun(
25002501
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
25012502
);
25022503
foreach (array_keys($parentRunsToProject) as $parentRunId) {
@@ -2507,7 +2508,7 @@ private function continueAsNew(
25072508
continue;
25082509
}
25092510

2510-
RunSummaryProjector::project(
2511+
$this->projectRun(
25112512
$parentRun->fresh([
25122513
'instance',
25132514
'tasks',
@@ -2521,7 +2522,7 @@ private function continueAsNew(
25212522
])
25222523
);
25232524
}
2524-
RunSummaryProjector::project(
2525+
$this->projectRun(
25252526
$continuedRun->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
25262527
);
25272528

@@ -2611,7 +2612,7 @@ private function completeRun(WorkflowRun $run, WorkflowTask $task, mixed $result
26112612

26122613
$this->dispatchParentResumeTasks($run);
26132614

2614-
RunSummaryProjector::project(
2615+
$this->projectRun(
26152616
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
26162617
);
26172618
}
@@ -2799,7 +2800,7 @@ private function timeoutRun(WorkflowRun $run, WorkflowTask $task): void
27992800
// Notify parent workflows and project summary.
28002801
$this->dispatchParentResumeTasks($run);
28012802

2802-
RunSummaryProjector::project(
2803+
$this->projectRun(
28032804
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
28042805
);
28052806
}
@@ -2899,7 +2900,7 @@ private function failRun(
28992900

29002901
$this->dispatchParentResumeTasks($run);
29012902

2902-
RunSummaryProjector::project(
2903+
$this->projectRun(
29032904
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
29042905
);
29052906
}
@@ -2978,7 +2979,7 @@ private function blockReplayUntilFailureCanBeRestored(
29782979
'lease_expires_at' => null,
29792980
])->save();
29802981

2981-
RunSummaryProjector::project(
2982+
$this->projectRun(
29822983
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
29832984
);
29842985
}
@@ -3008,7 +3009,7 @@ private function blockReplayUntilCompatibleConditionWaitDefinition(
30083009
'lease_expires_at' => null,
30093010
])->save();
30103011

3011-
RunSummaryProjector::project(
3012+
$this->projectRun(
30123013
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
30133014
);
30143015
}
@@ -3032,7 +3033,7 @@ private function blockReplayUntilCompatibleHistoryShape(
30323033
'lease_expires_at' => null,
30333034
])->save();
30343035

3035-
RunSummaryProjector::project(
3036+
$this->projectRun(
30363037
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
30373038
);
30383039
}
@@ -3365,7 +3366,7 @@ private function startChildRetryIfAvailable(
33653366

33663367
TaskDispatcher::dispatch($retryTask);
33673368

3368-
RunSummaryProjector::project(
3369+
$this->projectRun(
33693370
$retryRun->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
33703371
);
33713372

@@ -3439,7 +3440,7 @@ private function dispatchParentResumeTasks(WorkflowRun $childRun): void
34393440
$parentReference['parent_sequence'],
34403441
$childRun,
34413442
) !== null) {
3442-
RunSummaryProjector::project(
3443+
$this->projectRun(
34433444
$parentRun->fresh([
34443445
'instance',
34453446
'tasks',
@@ -3471,7 +3472,7 @@ private function dispatchParentResumeTasks(WorkflowRun $childRun): void
34713472
$childStatus
34723473
)
34733474
) {
3474-
RunSummaryProjector::project(
3475+
$this->projectRun(
34753476
$parentRun->fresh([
34763477
'instance',
34773478
'tasks',
@@ -3496,7 +3497,7 @@ private function dispatchParentResumeTasks(WorkflowRun $childRun): void
34963497
$parentReference['parent_sequence'],
34973498
)
34983499
) {
3499-
RunSummaryProjector::project(
3500+
$this->projectRun(
35003501
$parentRun->fresh([
35013502
'instance',
35023503
'tasks',
@@ -3525,7 +3526,7 @@ private function dispatchParentResumeTasks(WorkflowRun $childRun): void
35253526
WorkflowStepHistory::CHILD_WORKFLOW,
35263527
);
35273528
} catch (HistoryEventShapeMismatchException) {
3528-
RunSummaryProjector::project(
3529+
$this->projectRun(
35293530
$parentRun->fresh([
35303531
'instance',
35313532
'tasks',
@@ -3576,7 +3577,7 @@ private function dispatchParentResumeTasks(WorkflowRun $childRun): void
35763577

35773578
TaskDispatcher::dispatch($parentTask);
35783579

3579-
RunSummaryProjector::project(
3580+
$this->projectRun(
35803581
$parentRun->fresh([
35813582
'instance',
35823583
'tasks',
@@ -4343,7 +4344,7 @@ private function restartAfterPendingUpdateFailure(WorkflowRun $run, WorkflowTask
43434344
'compatibility' => $run->compatibility,
43444345
]);
43454346

4346-
RunSummaryProjector::project(
4347+
$this->projectRun(
43474348
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
43484349
);
43494350

@@ -4589,6 +4590,20 @@ private function visibleSequenceForAllCall(WorkflowRun $run, AllCall $current, i
45894590
return $sequence;
45904591
}
45914592

4593+
private function projectRun(WorkflowRun $run): void
4594+
{
4595+
$this->historyProjectionRole()
4596+
->projectRun($run);
4597+
}
4598+
4599+
private function historyProjectionRole(): HistoryProjectionRole
4600+
{
4601+
/** @var HistoryProjectionRole $role */
4602+
$role = app(HistoryProjectionRole::class);
4603+
4604+
return $role;
4605+
}
4606+
45924607
private static function intValue(mixed $value): ?int
45934608
{
45944609
if (is_int($value)) {

0 commit comments

Comments
 (0)