Skip to content

Commit c661862

Browse files
durable-workflow-opsdurable-workflow-ops
committed
Resolve external executor activity mappings
1 parent 1de632e commit c661862

6 files changed

Lines changed: 314 additions & 77 deletions

File tree

‎README.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -627,6 +627,10 @@ Handler mappings are config-first: set `DW_EXTERNAL_EXECUTOR_CONFIG_PATH` to a
627627
`DW_EXTERNAL_EXECUTOR_CONFIG_OVERLAY` to apply an environment overlay before
628628
server validation. Cluster discovery publishes the config contract and redacted
629629
runtime diagnostics at `worker_protocol.external_executor_config_contract`.
630+
When a leased activity task matches a valid configured activity mapping by task
631+
queue and activity type, the activity poll response includes a redacted
632+
`task.external_executor` mapping block with the handler, carrier target, auth
633+
reference, rollout metadata, and config schema version.
630634

631635
The carrier-neutral external task input envelope is published from
632636
`GET /api/cluster/info` at `worker_protocol.external_task_input_contract`.

‎app/Http/Controllers/Api/ActivityTaskController.php‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
use App\Models\WorkerRegistration;
66
use App\Support\ActivityTaskPoller;
7+
use App\Support\ExternalExecutorConfigContract;
78
use App\Support\NamespaceWorkflowScope;
89
use App\Support\WorkerProtocol;
910
use Illuminate\Http\JsonResponse;
@@ -85,6 +86,10 @@ public function poll(Request $request): JsonResponse
8586
'lease_owner' => $claim['lease_owner'],
8687
'lease_expires_at' => $claim['lease_expires_at'],
8788
'deadlines' => $this->executionDeadlines($claim['activity_execution_id'] ?? null),
89+
'external_executor' => ExternalExecutorConfigContract::resolveActivityMapping(
90+
(string) $claim['queue'],
91+
(string) $claim['activity_type'],
92+
),
8893
], static fn (mixed $v): bool => $v !== null),
8994
]);
9095
}

‎app/Support/ExternalExecutorConfigContract.php‎

Lines changed: 180 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ public static function manifest(): array
3939
'config_path_env' => 'DW_EXTERNAL_EXECUTOR_CONFIG_PATH',
4040
'overlay_env' => 'DW_EXTERNAL_EXECUTOR_CONFIG_OVERLAY',
4141
'cluster_info_path' => 'worker_protocol.external_executor_config_contract.runtime',
42-
'execution_status' => 'validation_and_discovery_only',
42+
'execution_status' => 'validation_discovery_and_activity_poll_resolution',
4343
],
4444
'validation' => [
4545
'fail_closed' => true,
@@ -57,92 +57,67 @@ public static function manifest(): array
5757
*/
5858
public static function runtime(): array
5959
{
60-
$path = self::configuredPath();
60+
return self::runtimeState()['runtime'];
61+
}
6162

62-
if ($path === null) {
63-
return [
64-
'configured' => false,
65-
'status' => 'not_configured',
66-
'source' => null,
67-
'overlay' => self::configuredOverlay(),
68-
'summary' => self::emptySummary(),
69-
'errors' => [],
70-
];
63+
/**
64+
* @return array<string, mixed>|null
65+
*/
66+
public static function resolveActivityMapping(string $taskQueue, string $activityType): ?array
67+
{
68+
$state = self::runtimeState();
69+
if ($state['runtime']['status'] !== 'valid' || ! is_array($state['document'])) {
70+
return null;
7171
}
7272

73-
$source = self::sourceInfo($path);
73+
/** @var array<string, mixed> $document */
74+
$document = $state['document'];
75+
$defaults = is_array($document['defaults'] ?? null) ? $document['defaults'] : [];
76+
$carriers = is_array($document['carriers'] ?? null) ? $document['carriers'] : [];
77+
$authRefs = is_array($document['auth_refs'] ?? null) ? $document['auth_refs'] : [];
78+
$mappings = is_array($document['mappings'] ?? null) ? $document['mappings'] : [];
7479

75-
if (! is_file($path) || ! is_readable($path)) {
76-
return [
77-
'configured' => true,
78-
'status' => 'invalid',
79-
'source' => $source,
80-
'overlay' => self::configuredOverlay(),
81-
'summary' => self::emptySummary(),
82-
'errors' => [
83-
self::error('unreadable_config', 'Configured external executor config file is not readable.'),
84-
],
85-
];
86-
}
80+
foreach ($mappings as $mapping) {
81+
if (! is_array($mapping) || self::stringValue($mapping['kind'] ?? null) !== 'activity') {
82+
continue;
83+
}
8784

88-
$contents = file_get_contents($path);
89-
if (! is_string($contents)) {
90-
return [
91-
'configured' => true,
92-
'status' => 'invalid',
93-
'source' => $source,
94-
'overlay' => self::configuredOverlay(),
95-
'summary' => self::emptySummary(),
96-
'errors' => [
97-
self::error('unreadable_config', 'Configured external executor config file could not be read.'),
98-
],
99-
];
100-
}
85+
$mappingQueue = self::stringValue($mapping['task_queue'] ?? $defaults['task_queue'] ?? null);
86+
$mappingActivityType = self::stringValue($mapping['activity_type'] ?? null);
10187

102-
try {
103-
$document = json_decode($contents, true, 512, JSON_THROW_ON_ERROR);
104-
} catch (\JsonException $exception) {
105-
return [
106-
'configured' => true,
107-
'status' => 'invalid',
108-
'source' => $source,
109-
'overlay' => self::configuredOverlay(),
110-
'summary' => self::emptySummary(),
111-
'errors' => [
112-
self::error('invalid_json', 'Configured external executor config is not valid JSON.', [
113-
'detail' => $exception->getMessage(),
114-
]),
115-
],
116-
];
117-
}
88+
if ($mappingQueue !== $taskQueue || $mappingActivityType !== $activityType) {
89+
continue;
90+
}
11891

119-
if (! is_array($document)) {
120-
return [
121-
'configured' => true,
122-
'status' => 'invalid',
123-
'source' => $source,
124-
'overlay' => self::configuredOverlay(),
125-
'summary' => self::emptySummary(),
126-
'errors' => [
127-
self::error('invalid_schema', 'Configured external executor config must be a JSON object.'),
128-
],
129-
];
130-
}
92+
$carrierName = self::stringValue($mapping['carrier'] ?? null);
93+
$carrier = $carrierName !== null && is_array($carriers[$carrierName] ?? null)
94+
? $carriers[$carrierName]
95+
: [];
96+
$authRef = self::stringValue($mapping['auth_ref'] ?? $defaults['auth_ref'] ?? null);
13197

132-
[$effective, $overlayError] = self::applyOverlay($document);
133-
$errors = self::validate($effective);
134-
if ($overlayError !== null) {
135-
array_unshift($errors, $overlayError);
98+
return array_filter([
99+
'schema' => self::CONFIG_SCHEMA.'.mapping',
100+
'version' => self::CONFIG_VERSION,
101+
'name' => self::stringValue($mapping['name'] ?? null),
102+
'kind' => 'activity',
103+
'activity_type' => $mappingActivityType,
104+
'task_queue' => $mappingQueue,
105+
'handler' => self::stringValue($mapping['handler'] ?? null),
106+
'carrier' => self::resolvedCarrier($carrierName, $carrier),
107+
'auth_ref' => $authRef,
108+
'auth' => $authRef !== null && is_array($authRefs[$authRef] ?? null)
109+
? self::redactValue($authRefs[$authRef])
110+
: null,
111+
'rollout' => is_array($mapping['rollout'] ?? null)
112+
? self::redactValue($mapping['rollout'])
113+
: null,
114+
'metadata' => is_array($mapping['metadata'] ?? null)
115+
? self::redactValue($mapping['metadata'])
116+
: null,
117+
], static fn (mixed $value): bool => $value !== null);
136118
}
137119

138-
return [
139-
'configured' => true,
140-
'status' => $errors === [] ? 'valid' : 'invalid',
141-
'source' => $source,
142-
'overlay' => self::configuredOverlay(),
143-
'summary' => self::summary($effective),
144-
'errors' => $errors,
145-
];
120+
return null;
146121
}
147122

148123
/**
@@ -210,6 +185,114 @@ private static function applyOverlay(array $document): array
210185
return [$document, null];
211186
}
212187

188+
/**
189+
* @return array{runtime: array<string, mixed>, document: array<string, mixed>|null}
190+
*/
191+
private static function runtimeState(): array
192+
{
193+
$path = self::configuredPath();
194+
195+
if ($path === null) {
196+
return [
197+
'runtime' => [
198+
'configured' => false,
199+
'status' => 'not_configured',
200+
'source' => null,
201+
'overlay' => self::configuredOverlay(),
202+
'summary' => self::emptySummary(),
203+
'errors' => [],
204+
],
205+
'document' => null,
206+
];
207+
}
208+
209+
$source = self::sourceInfo($path);
210+
211+
if (! is_file($path) || ! is_readable($path)) {
212+
return self::invalidRuntimeState($source, [
213+
self::error('unreadable_config', 'Configured external executor config file is not readable.'),
214+
]);
215+
}
216+
217+
$contents = file_get_contents($path);
218+
if (! is_string($contents)) {
219+
return self::invalidRuntimeState($source, [
220+
self::error('unreadable_config', 'Configured external executor config file could not be read.'),
221+
]);
222+
}
223+
224+
try {
225+
$document = json_decode($contents, true, 512, JSON_THROW_ON_ERROR);
226+
} catch (\JsonException $exception) {
227+
return self::invalidRuntimeState($source, [
228+
self::error('invalid_json', 'Configured external executor config is not valid JSON.', [
229+
'detail' => $exception->getMessage(),
230+
]),
231+
]);
232+
}
233+
234+
if (! is_array($document)) {
235+
return self::invalidRuntimeState($source, [
236+
self::error('invalid_schema', 'Configured external executor config must be a JSON object.'),
237+
]);
238+
}
239+
240+
[$effective, $overlayError] = self::applyOverlay($document);
241+
$errors = self::validate($effective);
242+
if ($overlayError !== null) {
243+
array_unshift($errors, $overlayError);
244+
}
245+
246+
return [
247+
'runtime' => [
248+
'configured' => true,
249+
'status' => $errors === [] ? 'valid' : 'invalid',
250+
'source' => $source,
251+
'overlay' => self::configuredOverlay(),
252+
'summary' => self::summary($effective),
253+
'errors' => $errors,
254+
],
255+
'document' => $errors === [] ? $effective : null,
256+
];
257+
}
258+
259+
/**
260+
* @param array{type: string, basename: string, sha256: string} $source
261+
* @param list<array<string, mixed>> $errors
262+
* @return array{runtime: array<string, mixed>, document: null}
263+
*/
264+
private static function invalidRuntimeState(array $source, array $errors): array
265+
{
266+
return [
267+
'runtime' => [
268+
'configured' => true,
269+
'status' => 'invalid',
270+
'source' => $source,
271+
'overlay' => self::configuredOverlay(),
272+
'summary' => self::emptySummary(),
273+
'errors' => $errors,
274+
],
275+
'document' => null,
276+
];
277+
}
278+
279+
/**
280+
* @param array<string, mixed> $carrier
281+
* @return array<string, mixed>|null
282+
*/
283+
private static function resolvedCarrier(?string $name, array $carrier): ?array
284+
{
285+
if ($name === null || $carrier === []) {
286+
return null;
287+
}
288+
289+
return array_filter([
290+
'name' => $name,
291+
'type' => self::stringValue($carrier['type'] ?? null),
292+
'target' => self::redactValue($carrier),
293+
], static fn (mixed $value): bool => $value !== null);
294+
}
295+
213296
/**
214297
* @return list<array<string, mixed>>
215298
*/
@@ -441,6 +524,26 @@ private static function redactContext(array $context): array
441524
return $redacted;
442525
}
443526

527+
private static function redactValue(mixed $value): mixed
528+
{
529+
if (! is_array($value)) {
530+
return $value;
531+
}
532+
533+
$redacted = [];
534+
foreach ($value as $key => $item) {
535+
if (preg_match('/token|secret|authorization|signature/i', (string) $key) === 1) {
536+
$redacted[$key] = 'redacted';
537+
538+
continue;
539+
}
540+
541+
$redacted[$key] = self::redactValue($item);
542+
}
543+
544+
return $redacted;
545+
}
546+
444547
private static function configuredPath(): ?string
445548
{
446549
$path = config('server.external_executor.config_path');

‎app/Support/ExternalTaskInputContract.php‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,12 @@ private static function activityTaskEnvelope(): array
166166
'task_queue' => ['source' => 'task.task_queue', 'type' => 'string'],
167167
'handler' => ['source' => 'task.activity_type', 'type' => 'string'],
168168
'connection' => ['source' => 'task.connection', 'type' => 'string', 'nullable' => true],
169+
'external_executor' => [
170+
'source' => 'task.external_executor',
171+
'type' => 'object',
172+
'nullable' => true,
173+
'meaning' => 'Resolved config-first handler mapping when DW_EXTERNAL_EXECUTOR_CONFIG_PATH matches this activity task.',
174+
],
169175
'idempotency_key' => ['source' => 'task.activity_attempt_id', 'type' => 'string'],
170176
],
171177
'workflow_fields' => [

0 commit comments

Comments
 (0)