Skip to content

Commit c8901f8

Browse files
durable-workflow-opsdurable-workflow-ops
committed
Workflow updates conformance: expose update diagnostics in Waterline views (#302)
1 parent f119e37 commit c8901f8

2 files changed

Lines changed: 335 additions & 0 deletions

File tree

‎app/Console/WorkflowUpdatesConformanceCommand.php‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,8 +125,15 @@ public function handle(HttpKernel $kernel): int
125125
$outputPath = $this->optionString('output');
126126
$artifactVersions = $this->artifactVersions();
127127
$artifactSources = $this->artifactSources();
128+
$originalConfig = [
129+
'waterline.engine_source' => config('waterline.engine_source'),
130+
'waterline.allow_unauthenticated' => config('waterline.allow_unauthenticated'),
131+
];
128132

129133
try {
134+
config()->set('waterline.engine_source', 'v2');
135+
config()->set('waterline.allow_unauthenticated', true);
136+
130137
$publicEvidence = $inputPath === null ? [] : $this->readJsonFile($inputPath);
131138
$detailCapture = $this->providedCapture('selected_run_detail', $startedAt);
132139
$historyCapture = $this->providedCapture('selected_run_history_export', $startedAt);
@@ -227,6 +234,10 @@ public function handle(HttpKernel $kernel): int
227234
$this->error('Waterline workflow update diagnostics command failed before the normal report completed.');
228235

229236
return self::FAILURE;
237+
} finally {
238+
foreach ($originalConfig as $key => $value) {
239+
config()->set($key, $value);
240+
}
230241
}
231242
}
232243

‎tests/Feature/WorkflowUpdatesConformanceCommandTest.php‎

Lines changed: 324 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,20 @@
22

33
namespace Waterline\Tests\Feature;
44

5+
use Illuminate\Support\Str;
6+
use Waterline\Waterline;
57
use Waterline\Tests\TestCase;
68
use Workflow\Serializers\Serializer;
9+
use Workflow\V2\Contracts\HistoryExportRedactor;
10+
use Workflow\V2\Contracts\OperatorObservabilityRepository;
11+
use Workflow\V2\Enums\HistoryEventType;
12+
use Workflow\V2\Models\WorkflowCommand;
13+
use Workflow\V2\Models\WorkflowFailure;
14+
use Workflow\V2\Models\WorkflowHistoryEvent;
15+
use Workflow\V2\Models\WorkflowInstance;
16+
use Workflow\V2\Models\WorkflowRun;
17+
use Workflow\V2\Models\WorkflowRunSummary;
18+
use Workflow\V2\Models\WorkflowUpdate;
719

820
class WorkflowUpdatesConformanceCommandTest extends TestCase
921
{
@@ -68,6 +80,84 @@ public function testItEmitsSelectedRunUpdateDiagnosticsForProvidedCaptures(): vo
6880
}
6981
}
7082

83+
public function testItSelfCapturesSelectedRunUpdateDiagnosticsInPublishedHostMode(): void
84+
{
85+
$output = tempnam(sys_get_temp_dir(), 'waterline-wu-output-');
86+
$this->assertIsString($output);
87+
88+
config()->set('waterline.engine_source', 'auto');
89+
config()->set('waterline.allow_unauthenticated', false);
90+
Waterline::auth(fn (): bool => false);
91+
$fixture = $this->seedSelectedRunUpdateDiagnostics();
92+
93+
$this->app->bind(OperatorObservabilityRepository::class, fn (): OperatorObservabilityRepository => new class implements OperatorObservabilityRepository {
94+
public function runDetail(WorkflowRun $run, ?int $timelineLimit = null): array
95+
{
96+
throw new \RuntimeException('selected-run projection unavailable');
97+
}
98+
99+
public function listItem(WorkflowRunSummary $summary): array
100+
{
101+
return [];
102+
}
103+
104+
public function runHistoryExport(
105+
WorkflowRun $run,
106+
?\Carbon\CarbonInterface $exportedAt = null,
107+
HistoryExportRedactor|callable|null $redactor = null,
108+
): array {
109+
throw new \RuntimeException('history export projection unavailable');
110+
}
111+
112+
public function dashboardSummary(?\Carbon\CarbonInterface $now = null, ?string $namespace = null): array
113+
{
114+
return [];
115+
}
116+
117+
public function metrics(?\Carbon\CarbonInterface $now = null, ?string $namespace = null): array
118+
{
119+
return [];
120+
}
121+
});
122+
123+
try {
124+
$this->artisan('waterline:workflow-updates-conformance', [
125+
'--output' => $output,
126+
'--run-id' => 'operator-diagnostics-'.$fixture['run']->id,
127+
'--instance-id' => $fixture['instance']->id,
128+
'--workflow-run-id' => $fixture['run']->id,
129+
] + $this->publishedArtifactOptions())->assertExitCode(0);
130+
131+
$result = json_decode((string) file_get_contents($output), true, 512, JSON_THROW_ON_ERROR);
132+
$scenarios = array_column($result['scenario_results'], null, 'scenario_id');
133+
$scenario = $scenarios['operator_diagnostics_surfaces'];
134+
$captures = $scenario['observed_outputs']['api_captures'];
135+
$matrix = $scenario['observed_outputs']['operator_surface_matrix'];
136+
137+
$this->assertSame('pass', $scenario['status']);
138+
$this->assertSame(200, $captures['selected_run_detail']['status']);
139+
$this->assertSame(200, $captures['selected_run_history_export']['status']);
140+
$this->assertSame('published_waterline_artifact_http_route', $captures['selected_run_detail']['capture_source']);
141+
$this->assertSame(
142+
'/waterline/api/instances/'.$fixture['instance']->id.'/runs/'.$fixture['run']->id,
143+
$captures['selected_run_detail']['path'],
144+
);
145+
$this->assertSame(1, $matrix['state_counts']['accepted']);
146+
$this->assertSame(1, $matrix['state_counts']['completed']);
147+
$this->assertSame(1, $matrix['state_counts']['failed']);
148+
$this->assertSame(1, $matrix['state_counts']['refused']);
149+
$this->assertTrue($matrix['states']['accepted']['request_identifiers_visible']);
150+
$this->assertTrue($matrix['states']['completed']['result_visible']);
151+
$this->assertTrue($matrix['states']['failed']['error_visible']);
152+
$this->assertTrue($matrix['states']['refused']['history_export_references_visible']);
153+
$this->assertSame('auto', config('waterline.engine_source'));
154+
$this->assertFalse(config('waterline.allow_unauthenticated'));
155+
} finally {
156+
Waterline::auth(fn (): bool => true);
157+
$this->unlinkTemp($output);
158+
}
159+
}
160+
71161
public function testItFailsClosedWhenARequiredUpdatePathLacksErrorDiagnostics(): void
72162
{
73163
$output = tempnam(sys_get_temp_dir(), 'waterline-wu-output-');
@@ -256,6 +346,240 @@ private function publishedArtifactOptions(): array
256346
];
257347
}
258348

349+
/**
350+
* @return array{instance: WorkflowInstance, run: WorkflowRun}
351+
*/
352+
private function seedSelectedRunUpdateDiagnostics(): array
353+
{
354+
$instance = WorkflowInstance::create([
355+
'id' => 'update-conformance-instance',
356+
'workflow_class' => 'App\\Workflows\\WorkflowUpdatesConformanceWorkflow',
357+
'workflow_type' => 'workflow-updates.probe',
358+
'run_count' => 1,
359+
'started_at' => now()->subMinutes(5),
360+
]);
361+
362+
$run = WorkflowRun::create([
363+
'id' => (string) Str::ulid(),
364+
'workflow_instance_id' => $instance->id,
365+
'run_number' => 1,
366+
'workflow_class' => 'App\\Workflows\\WorkflowUpdatesConformanceWorkflow',
367+
'workflow_type' => 'workflow-updates.probe',
368+
'status' => 'waiting',
369+
'arguments' => Serializer::serialize(['operator-diagnostics']),
370+
'connection' => 'redis',
371+
'queue' => 'workflow-updates-operator-queue',
372+
'started_at' => now()->subMinutes(5),
373+
'last_progress_at' => now()->subMinute(),
374+
]);
375+
376+
$instance->update(['current_run_id' => $run->id]);
377+
378+
$accepted = $this->seedWorkflowUpdate($instance, $run, 1, 'accepted', 'approve', [true, 'cli-accepted']);
379+
WorkflowHistoryEvent::record($run, HistoryEventType::UpdateAccepted, [
380+
'workflow_command_id' => $accepted['command']->id,
381+
'update_id' => $accepted['update']->id,
382+
'workflow_instance_id' => $instance->id,
383+
'workflow_run_id' => $run->id,
384+
'update_name' => 'approve',
385+
'arguments' => Serializer::serialize([true, 'cli-accepted']),
386+
], null, $accepted['command']);
387+
388+
$completed = $this->seedWorkflowUpdate($instance, $run, 2, 'completed', 'approve', [true, 'cli-completed'], [
389+
'result' => ['approved' => true, 'source' => 'operator-diagnostics-cli-waterline'],
390+
'outcome' => 'update_completed',
391+
'closed_at' => now()->subSeconds(30),
392+
]);
393+
WorkflowHistoryEvent::record($run, HistoryEventType::UpdateAccepted, [
394+
'workflow_command_id' => $completed['command']->id,
395+
'update_id' => $completed['update']->id,
396+
'workflow_instance_id' => $instance->id,
397+
'workflow_run_id' => $run->id,
398+
'update_name' => 'approve',
399+
'arguments' => Serializer::serialize([true, 'cli-completed']),
400+
], null, $completed['command']);
401+
WorkflowHistoryEvent::record($run, HistoryEventType::UpdateCompleted, [
402+
'workflow_command_id' => $completed['command']->id,
403+
'update_id' => $completed['update']->id,
404+
'workflow_instance_id' => $instance->id,
405+
'workflow_run_id' => $run->id,
406+
'update_name' => 'approve',
407+
'result' => Serializer::serialize(['approved' => true, 'source' => 'operator-diagnostics-cli-waterline']),
408+
], null, $completed['command']);
409+
410+
$failed = $this->seedWorkflowUpdate($instance, $run, 3, 'failed', 'fail_update', ['cli failure'], [
411+
'outcome' => 'update_failed',
412+
'closed_at' => now()->subSeconds(20),
413+
]);
414+
$failure = WorkflowFailure::create([
415+
'id' => (string) Str::ulid(),
416+
'workflow_run_id' => $run->id,
417+
'source_kind' => 'workflow_command',
418+
'source_id' => $failed['command']->id,
419+
'propagation_kind' => 'update',
420+
'handled' => false,
421+
'exception_class' => 'DurableWorkflow\\Conformance\\WorkflowUpdateOperatorDiagnosticsFailure',
422+
'message' => 'workflow update operator diagnostics failure',
423+
'file' => __FILE__,
424+
'line' => 1,
425+
'trace_preview' => '',
426+
]);
427+
$failed['update']->update(['failure_id' => $failure->id]);
428+
WorkflowHistoryEvent::record($run, HistoryEventType::UpdateAccepted, [
429+
'workflow_command_id' => $failed['command']->id,
430+
'update_id' => $failed['update']->id,
431+
'workflow_instance_id' => $instance->id,
432+
'workflow_run_id' => $run->id,
433+
'update_name' => 'fail_update',
434+
'arguments' => Serializer::serialize(['cli failure']),
435+
], null, $failed['command']);
436+
WorkflowHistoryEvent::record($run, HistoryEventType::UpdateCompleted, [
437+
'workflow_command_id' => $failed['command']->id,
438+
'update_id' => $failed['update']->id,
439+
'workflow_instance_id' => $instance->id,
440+
'workflow_run_id' => $run->id,
441+
'update_name' => 'fail_update',
442+
'failure_id' => $failure->id,
443+
'exception_class' => 'DurableWorkflow\\Conformance\\WorkflowUpdateOperatorDiagnosticsFailure',
444+
'message' => 'workflow update operator diagnostics failure',
445+
], null, $failed['command']);
446+
447+
$refused = $this->seedWorkflowUpdate($instance, $run, 4, 'rejected', 'missing_update', [], [
448+
'outcome' => 'rejected_unknown_update',
449+
'rejection_reason' => 'unknown_update',
450+
'validation_errors' => ['update' => ['The requested workflow update is not declared.']],
451+
'closed_at' => now()->subSeconds(10),
452+
]);
453+
WorkflowHistoryEvent::record($run, HistoryEventType::UpdateRejected, [
454+
'workflow_command_id' => $refused['command']->id,
455+
'update_id' => $refused['update']->id,
456+
'workflow_instance_id' => $instance->id,
457+
'workflow_run_id' => $run->id,
458+
'update_name' => 'missing_update',
459+
'arguments' => Serializer::serialize([]),
460+
'rejection_reason' => 'unknown_update',
461+
'validation_errors' => ['update' => ['The requested workflow update is not declared.']],
462+
], null, $refused['command']);
463+
464+
WorkflowRunSummary::create([
465+
'id' => $run->id,
466+
'workflow_instance_id' => $instance->id,
467+
'run_number' => 1,
468+
'is_current_run' => true,
469+
'engine_source' => 'v2',
470+
'class' => 'App\\Workflows\\WorkflowUpdatesConformanceWorkflow',
471+
'workflow_type' => 'workflow-updates.probe',
472+
'status' => 'waiting',
473+
'status_bucket' => 'running',
474+
'connection' => 'redis',
475+
'queue' => 'workflow-updates-operator-queue',
476+
'history_event_count' => 6,
477+
'history_size_bytes' => 2048,
478+
'started_at' => $run->started_at,
479+
'created_at' => now()->subMinutes(5),
480+
'updated_at' => now()->subMinute(),
481+
]);
482+
483+
return ['instance' => $instance, 'run' => $run];
484+
}
485+
486+
/**
487+
* @return array{command: WorkflowCommand, update: WorkflowUpdate}
488+
*/
489+
private function seedWorkflowUpdate(
490+
WorkflowInstance $instance,
491+
WorkflowRun $run,
492+
int $sequence,
493+
string $status,
494+
string $name,
495+
array $arguments,
496+
array $overrides = [],
497+
): array {
498+
$outcome = $overrides['outcome'] ?? ($status === 'completed' ? 'update_completed' : null);
499+
$rejectionReason = $overrides['rejection_reason'] ?? null;
500+
$acceptedAt = now()->subSeconds(60 - $sequence);
501+
$closedAt = $overrides['closed_at'] ?? null;
502+
$commandStatus = $status === 'rejected' ? 'rejected' : 'accepted';
503+
504+
$command = WorkflowCommand::create([
505+
'id' => (string) Str::ulid(),
506+
'workflow_instance_id' => $instance->id,
507+
'workflow_run_id' => $run->id,
508+
'command_sequence' => $sequence,
509+
'command_type' => 'update',
510+
'target_scope' => 'instance',
511+
'source' => 'api',
512+
'context' => [
513+
'caller' => [
514+
'type' => 'operator',
515+
'label' => 'Operator API',
516+
],
517+
'principal' => [
518+
'type' => 'service-account',
519+
'id' => 'workflow-updates-operator',
520+
'label' => 'Workflow Updates Operator',
521+
],
522+
'auth' => [
523+
'status' => 'verified',
524+
'method' => 'bearer',
525+
],
526+
'request' => [
527+
'method' => 'POST',
528+
'path' => '/waterline/api/instances/'.$instance->id.'/runs/'.$run->id.'/updates/'.$name,
529+
'route_name' => 'waterline.instances.runs.update',
530+
'fingerprint' => 'sha256:update-'.$sequence,
531+
'request_id' => 'cli-'.$status.'-'.$sequence,
532+
'correlation_id' => 'corr-update-'.$sequence,
533+
],
534+
],
535+
'status' => $commandStatus,
536+
'outcome' => $outcome,
537+
'rejection_reason' => $rejectionReason,
538+
'workflow_class' => $run->workflow_class,
539+
'workflow_type' => $run->workflow_type,
540+
'payload_codec' => config('workflows.serializer'),
541+
'payload' => Serializer::serialize([
542+
'name' => $name,
543+
'arguments' => $arguments,
544+
]),
545+
'accepted_at' => $commandStatus === 'accepted' ? $acceptedAt : null,
546+
'rejected_at' => $commandStatus === 'rejected' ? $closedAt : null,
547+
'applied_at' => $status === 'completed' || $status === 'failed' ? $closedAt : null,
548+
'created_at' => $acceptedAt,
549+
'updated_at' => $closedAt ?? $acceptedAt,
550+
]);
551+
552+
$update = WorkflowUpdate::create([
553+
'id' => (string) Str::ulid(),
554+
'workflow_command_id' => $command->id,
555+
'workflow_instance_id' => $instance->id,
556+
'workflow_run_id' => $run->id,
557+
'target_scope' => 'instance',
558+
'requested_workflow_run_id' => null,
559+
'resolved_workflow_run_id' => $run->id,
560+
'update_name' => $name,
561+
'status' => $status,
562+
'outcome' => $outcome,
563+
'rejection_reason' => $rejectionReason,
564+
'validation_errors' => $overrides['validation_errors'] ?? [],
565+
'command_sequence' => $sequence,
566+
'workflow_sequence' => $status === 'accepted' || $status === 'rejected' ? null : $sequence - 1,
567+
'payload_codec' => config('workflows.serializer'),
568+
'arguments' => Serializer::serialize($arguments),
569+
'result' => array_key_exists('result', $overrides)
570+
? Serializer::serialize($overrides['result'])
571+
: null,
572+
'accepted_at' => $commandStatus === 'accepted' ? $acceptedAt : null,
573+
'applied_at' => $status === 'completed' || $status === 'failed' ? $closedAt : null,
574+
'rejected_at' => $commandStatus === 'rejected' ? $closedAt : null,
575+
'closed_at' => $closedAt,
576+
'created_at' => $acceptedAt,
577+
'updated_at' => $closedAt ?? $acceptedAt,
578+
]);
579+
580+
return ['command' => $command, 'update' => $update];
581+
}
582+
259583
/**
260584
* @return array<string, mixed>
261585
*/

0 commit comments

Comments
 (0)