From c44c6351eb64c88f2471d27ea2565795695c8e80 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 16:44:14 -0600 Subject: [PATCH] Prepare capability-gated worker dispatch latency observations --- CHANGELOG.md | 4 + docs/portal/operate/metrics.md | 31 ++- resources/protocol/sdk-metrics-v1.json | 10 +- src/Client.php | 24 +- src/Metrics/MetricsContract.php | 22 +- src/Metrics/MetricsObserver.php | 16 ++ src/Worker.php | 21 ++ tests/ClientMetricsTest.php | 2 +- tests/MetricsRecorderAdaptersTest.php | 36 +++ tests/WorkerDispatchMetricsTest.php | 310 +++++++++++++++++++++++++ 10 files changed, 469 insertions(+), 7 deletions(-) create mode 100644 tests/WorkerDispatchMetricsTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index b25f73b2..a4da8b7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ project follows [Semantic Versioning](https://semver.org/). ### Added +- An unreleased schedule-to-start histogram for workflow/activity dispatch uses + the runtime's capability-gated `ready_age_ms` claim metadata (shared metrics + revision 2, central #233). Valid zero is retained; missing, unsupported or + malformed ages are omitted. Existing instruments keep their names and tags. - `WorkflowContext::startChildWorkflow()` starts a child and returns a `ChildWorkflowHandle` once the runtime has committed the start (#185). The handle's `identity()` gives the child call ID, the child workflow ID and the diff --git a/docs/portal/operate/metrics.md b/docs/portal/operate/metrics.md index f1940977..7a91dc29 100644 --- a/docs/portal/operate/metrics.md +++ b/docs/portal/operate/metrics.md @@ -56,14 +56,43 @@ result, a retry decision, a deadline or task authority. | `durable_workflow_worker_poll_duration_seconds` | histogram | `s` | same as above | | `durable_workflow_worker_tasks` | counter | `1` | `task_kind`, `task_queue`, `outcome` | | `durable_workflow_worker_task_duration_seconds` | histogram | `s` | same as above | +| `durable_workflow_worker_task_schedule_to_start_seconds` (unreleased) | histogram | `s` | `task_kind`, `task_queue` | -Each observation increments its counter by one and records one nonnegative +Each request, poll or processing observation increments its counter by one and records one nonnegative duration with the same tags. A client observation covers one logical request, including the SDK's internal transport retries. Poll and task observations describe this worker process, not durable execution counts, workflow completion or schedule-to-start latency. A missing observation does not mean zero work or a healthy process. +## Schedule-to-start timing (unreleased) + +The revision-2 dispatch histogram records only workflow and activity tasks that +this worker actually dispatches. The runtime must explicitly advertise Boolean +`claimed_task_ready_age: true` in discovery or registration capabilities and +provide an integer `ready_age_ms` between `0` and `9007199254740991` on the claim. +The recorder receives that value divided by 1000, with only `task_kind` and +`task_queue` tags. A real zero is recorded. There is no paired counter or +outcome tag. + +The Server measures from the attempt's eligibility to its committed lease. +Scheduled delay and retry backoff are excluded. Missing or unsupported metadata +produces no sample. Malformed metadata produces no sample and a fixed diagnostic +without its value. The worker never substitutes a local clock difference, an +old creation time or zero for an unknown origin. + +Managed registration caches its capability response. When using `tick()` alone, +discover with `$client->workerClusterInfo()` before constructing the worker. +Dispatch adds no discovery request. Query work, empty or refused polls and HTTP +retries produce no schedule-to-start sample. A replayed claim retains its +original age; actually dispatching it again is another process observation, +so this histogram does not establish exactly-once execution. + +The vendored registry pins shared contract revision 2 at +[`23f9bc3c`](https://github.com/durable-workflow/.github/tree/23f9bc3c3798cb6aab6ccf5f9cd210854b820c65/sdk-metrics). +Server eligibility-origin implementation and exact published Server/installed +SDK qualification remain required before this addition is delivered. + ## Bounded labels Labels never carry identifiers or secrets: diff --git a/resources/protocol/sdk-metrics-v1.json b/resources/protocol/sdk-metrics-v1.json index 07263924..f6b675ef 100644 --- a/resources/protocol/sdk-metrics-v1.json +++ b/resources/protocol/sdk-metrics-v1.json @@ -1,6 +1,6 @@ { "schema": "durable-workflow.sdk-metrics/v1", - "revision": 1, + "revision": 2, "reference": { "repository": "durable-workflow/sdk-python", "commit": "2da02a7ac8a68a7cd4267e86ef10e0198a309592", @@ -58,6 +58,14 @@ "recorder_method": "record", "scope": "worker_task", "tags": ["task_kind", "task_queue", "outcome"] + }, + { + "name": "durable_workflow_worker_task_schedule_to_start_seconds", + "type": "histogram", + "unit": "s", + "recorder_method": "record", + "scope": "worker_task_dispatch", + "tags": ["task_kind", "task_queue"] } ] } diff --git a/src/Client.php b/src/Client.php index 9d27ddb5..5adb134a 100644 --- a/src/Client.php +++ b/src/Client.php @@ -89,6 +89,7 @@ final class Client implements WorkflowClientInterface private ?string $activityLeaseBootstrap = null; private ?ExecutionTelemetry $executionTelemetry = null; private bool $observabilityContextSupported = false; + private bool $claimedTaskReadyAgeSupported = false; private MetricsObserver $metricsObserver; private ?ClientRequestObservation $metricsRequest = null; /** @var (\Closure(string, array|null, array, RequestBudget|null): (?\Closure))|null */ @@ -335,10 +336,12 @@ public function health(): array public function clusterInfo(bool $includeDiagnostics = false): ClusterInfo { $this->observabilityContextSupported = false; + $this->claimedTaskReadyAgeSupported = false; $path = $includeDiagnostics ? self::CLUSTER_INFO_PATH.'?include=diagnostics' : self::CLUSTER_INFO_PATH; $info = $this->control('GET', $path); $this->observabilityContextSupported = TraceCarrier::supported($info); + $this->rememberClaimedTaskReadyAge($info['worker_protocol']['server_capabilities'] ?? null); return ClusterInfo::fromArray($info); } @@ -449,12 +452,26 @@ private function serviceCatalogRead(string $path, array $filters = [], ?string $ public function workerClusterInfo(): ClusterInfo { $this->observabilityContextSupported = false; + $this->claimedTaskReadyAgeSupported = false; $info = $this->worker('GET', self::CLUSTER_INFO_PATH); $this->observabilityContextSupported = TraceCarrier::supported($info); + $this->rememberClaimedTaskReadyAge($info['worker_protocol']['server_capabilities'] ?? null); return ClusterInfo::fromArray($info); } + /** @internal Read cached discovery without adding I/O to task dispatch. */ + public function claimedTaskReadyAgeAvailable(): bool + { + return $this->claimedTaskReadyAgeSupported; + } + + private function rememberClaimedTaskReadyAge(mixed $capabilities): void + { + $this->claimedTaskReadyAgeSupported = is_array($capabilities) + && ($capabilities['claimed_task_ready_age'] ?? null) === true; + } + /** * @param list $input * @param array|null $memo @@ -1635,7 +1652,7 @@ public function registerWorker( } } - return $this->worker('POST', '/worker/register', $this->withoutNulls([ + $registration = $this->worker('POST', '/worker/register', $this->withoutNulls([ 'worker_id' => $workerId, 'task_queue' => $taskQueue, 'runtime' => 'php', @@ -1651,6 +1668,11 @@ public function registerWorker( 'max_concurrent_worker_sessions' => $maxConcurrentWorkerSessions, 'build_id' => $buildId, ])); + if (array_key_exists('server_capabilities', $registration)) { + $this->rememberClaimedTaskReadyAge($registration['server_capabilities']); + } + + return $registration; } /** @return array */ diff --git a/src/Metrics/MetricsContract.php b/src/Metrics/MetricsContract.php index abd430fc..0251f742 100644 --- a/src/Metrics/MetricsContract.php +++ b/src/Metrics/MetricsContract.php @@ -7,7 +7,7 @@ /** @internal Loads the exact pinned shared registry, not a PHP-specific metric list. */ final class MetricsContract { - /** @var array|null */ + /** @var array|null */ private static ?array $pairs = null; /** @return array{counter: string, histogram: string} */ @@ -15,10 +15,22 @@ public static function pair(string $scope): array { self::$pairs ??= self::load(); - return self::$pairs[$scope] ?? throw new \InvalidArgumentException('Unknown SDK metrics scope.'); + $pair = self::$pairs[$scope] ?? null; + if (!isset($pair['counter'], $pair['histogram'])) { + throw new \InvalidArgumentException('SDK metrics scope has no counter/histogram pair.'); + } + + return ['counter' => $pair['counter'], 'histogram' => $pair['histogram']]; + } + + public static function histogram(string $scope): string + { + self::$pairs ??= self::load(); + + return self::$pairs[$scope]['histogram'] ?? throw new \InvalidArgumentException('Unknown SDK metrics scope.'); } - /** @return array */ + /** @return array */ private static function load(): array { $document = file_get_contents(dirname(__DIR__, 2).'/resources/protocol/sdk-metrics-v1.json'); @@ -37,6 +49,10 @@ private static function load(): array } $complete = []; foreach ($pairs as $scope => $pair) { + if ($scope === 'worker_task_dispatch' && isset($pair['histogram']) && !isset($pair['counter'])) { + $complete[$scope] = ['histogram' => $pair['histogram']]; + continue; + } if (!isset($pair['counter'], $pair['histogram'])) { throw new InvalidMetricsRegistry('The SDK metrics registry has an incomplete scope.'); } diff --git a/src/Metrics/MetricsObserver.php b/src/Metrics/MetricsObserver.php index 0d6d4011..ed82a3dc 100644 --- a/src/Metrics/MetricsObserver.php +++ b/src/Metrics/MetricsObserver.php @@ -50,4 +50,20 @@ public function observe(string $scope, float $seconds, array $tags): void $this->recording = false; } } + + /** @param array $tags */ + public function observeHistogram(string $scope, float $seconds, array $tags): void + { + if (!$this->enabled() || !is_finite($seconds) || $seconds < 0) { + return; + } + $this->recording = true; + try { + $this->recorder->record(MetricsContract::histogram($scope), $seconds, $tags); + } catch (Throwable) { + // Missing telemetry never authorizes repeating application work. + } finally { + $this->recording = false; + } + } } diff --git a/src/Worker.php b/src/Worker.php index e40f067c..a22257cb 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -1236,6 +1236,7 @@ private function taskFromPoll(array $response): ?array /** @param array $task */ private function executePolledTask(string $taskKind, array $task): void { + $this->observeTaskReadyAge($taskKind, $task); if (!$this->metricsObserver->enabled()) { $this->dispatchPolledTask($taskKind, $task); @@ -1255,6 +1256,26 @@ private function executePolledTask(string $taskKind, array $task): void } } + /** @param array $task */ + private function observeTaskReadyAge(string $taskKind, array $task): void + { + if (!$this->metricsObserver->enabled() || !in_array($taskKind, ['workflow', 'activity'], true) + || !$this->client->claimedTaskReadyAgeAvailable() || !array_key_exists('ready_age_ms', $task)) { + return; + } + $milliseconds = $task['ready_age_ms']; + if (!is_int($milliseconds) || $milliseconds < 0 || $milliseconds > 9007199254740991) { + try { + $this->diagnostic('worker.metrics_invalid_ready_age', ['task_kind' => $taskKind], 'warning'); + } catch (Throwable) { + // A diagnostic failure cannot change the claimed task's result. + } + return; + } + $this->metricsObserver->observeHistogram('worker_task_dispatch', $milliseconds / 1000, + ['task_kind' => $taskKind, 'task_queue' => $this->taskQueue]); + } + /** @param array $task */ private function dispatchPolledTask(string $taskKind, array $task): void { diff --git a/tests/ClientMetricsTest.php b/tests/ClientMetricsTest.php index dc6027f2..3e1a644f 100644 --- a/tests/ClientMetricsTest.php +++ b/tests/ClientMetricsTest.php @@ -17,7 +17,7 @@ final class ClientMetricsTest extends TestCase { public function testSharedContractBytesStayPinned(): void { - self::assertSame('c456f4f434c601fa26f0caba07ef0fd29f548ca8f5fc29ba2f6430f01e32c0f0', + self::assertSame('081d98ec6f5f6e3812f81ddbe8d2589d75e7c6e457c580dc6631d7857594516a', hash_file('sha256', dirname(__DIR__).'/resources/protocol/sdk-metrics-v1.json')); } diff --git a/tests/MetricsRecorderAdaptersTest.php b/tests/MetricsRecorderAdaptersTest.php index cf639d90..c54073de 100644 --- a/tests/MetricsRecorderAdaptersTest.php +++ b/tests/MetricsRecorderAdaptersTest.php @@ -8,6 +8,8 @@ use DurableWorkflow\Metrics\InMemoryMetricsRecorder; use DurableWorkflow\Metrics\OpenTelemetryMetricsRecorder; use DurableWorkflow\Tests\Support\FakeTransport; +use DurableWorkflow\Worker; +use DurableWorkflow\Worker\ActivityContext; use OpenTelemetry\SDK\Metrics\Data\Histogram; use OpenTelemetry\SDK\Metrics\Data\Sum; use OpenTelemetry\SDK\Metrics\MeterProvider; @@ -76,4 +78,38 @@ public function testOpenTelemetryRecorderExportsContractNamesUnitsAndBoundedAttr self::assertStringNotContainsString('private-', serialize(array_map( static fn ($metric): array => [$metric->name, $metric->unit], $metrics))); } + + public function testActualActivityDispatchExportsTheScheduleToStartHistogram(): void + { + $exporter = new InMemoryExporter(); + $reader = new ExportingReader($exporter); + $provider = MeterProvider::builder()->addReader($reader)->build(); + $client = new Client('https://server.invalid', metrics: new OpenTelemetryMetricsRecorder($provider), + transport: new FakeTransport([ + ['worker_protocol' => ['server_capabilities' => ['claimed_task_ready_age' => true]]], + ['task' => null, 'poll_status' => 'empty'], + ['task' => ['task_id' => 'activity-1', 'activity_attempt_id' => 'attempt-1', + 'lease_owner' => 'worker-1', 'activity_type' => 'orders.simple', 'payload_codec' => 'avro', 'ready_age_ms' => 1500], + 'poll_status' => 'leased'], + ['recorded' => true, 'outcome' => 'completed'], + ['task' => null, 'poll_status' => 'empty'], + ])); + $client->workerClusterInfo(); + $worker = new Worker($client, 'orders', workerId: 'worker-1', clock: static fn (): float => 0); + $worker->registerActivity('orders.simple', static fn (ActivityContext $context): string => 'done'); + self::assertTrue($worker->tick(0)); + $reader->collect(); + $metrics = []; + foreach ($exporter->collect() as $metric) { + $metrics[$metric->name] = $metric; + } + $metric = $metrics['durable_workflow_worker_task_schedule_to_start_seconds']; + self::assertSame('s', $metric->unit); + self::assertInstanceOf(Histogram::class, $metric->data); + $points = [...$metric->data->dataPoints]; + self::assertCount(1, $points); + self::assertSame(1, $points[0]->count); + self::assertSame(1.5, $points[0]->sum); + self::assertSame(['task_kind' => 'activity', 'task_queue' => 'orders'], $points[0]->attributes->toArray()); + } } diff --git a/tests/WorkerDispatchMetricsTest.php b/tests/WorkerDispatchMetricsTest.php new file mode 100644 index 00000000..733dd0d0 --- /dev/null +++ b/tests/WorkerDispatchMetricsTest.php @@ -0,0 +1,310 @@ +worker($recorder, $kind, ['ready_age_ms' => $age]); + $called = 0; + $atDispatch = []; + $handler = function () use (&$called, &$atDispatch, $recorder): string { + ++$called; + $atDispatch = self::samples($recorder); + + return 'done'; + }; + if ($kind === 'activity') { + $worker->registerActivity('orders.simple', static fn (ActivityContext $context): string => $handler()); + } else { + $worker->registerWorkflow('orders.simple', static fn (WorkflowContext $context): string => $handler()); + } + self::assertTrue($worker->tick(0)); + self::assertSame(1, $called); + self::assertSame([['method' => 'record', 'name' => self::METRIC, 'value' => (float) $age / 1000, + 'tags' => ['task_kind' => $kind, 'task_queue' => 'orders']]], $atDispatch); + self::assertCount(1, self::samples($recorder)); + self::assertCount(1, array_filter($transport->requests, static fn (array $r): bool => str_ends_with($r['uri'], '/cluster/info'))); + self::assertStringNotContainsString('private-', json_encode($recorder->events, JSON_THROW_ON_ERROR)); + } + + /** @return array */ + public static function validAges(): array + { + $cases = []; + foreach (['workflow', 'activity'] as $kind) { + foreach ([0, 1, 1500, 9007199254740991] as $age) { + $cases[$kind.' '.$age] = [$kind, $age]; + } + } + + return $cases; + } + + #[DataProvider('invalidAges')] + public function testInvalidAgeProducesOnlyABoundedDiagnostic(mixed $age): void + { + $recorder = new RecordingMetricsRecorder(); + $diagnostics = []; + [$worker, $client] = $this->worker($recorder, 'activity', ['ready_age_ms' => $age], diagnostics: $diagnostics); + $worker->registerActivity('orders.simple', static fn (ActivityContext $context): string => 'done'); + self::assertTrue($worker->tick(0)); + self::assertSame([], self::samples($recorder)); + $invalid = array_values(array_filter($diagnostics, static fn (array $row): bool => $row[0] === 'worker.metrics_invalid_ready_age')); + self::assertSame([['worker.metrics_invalid_ready_age', ['task_kind' => 'activity']]], $invalid); + } + + /** @return array */ + public static function invalidAges(): array + { + return ['null' => [null], 'true' => [true], 'false' => [false], 'text' => ['1500'], + 'integral float' => [1500.0], 'negative' => [-1], 'overflow' => [9007199254740992], + 'infinity' => [INF], 'nan' => [NAN], 'array' => [[]]]; + } + + #[DataProvider('unsupportedCapabilities')] + public function testUnsupportedCapabilityNeverTrustsTaskMetadata(mixed $capability): void + { + $recorder = new RecordingMetricsRecorder(); + [$worker, $client] = $this->worker($recorder, 'activity', ['ready_age_ms' => 1500], $capability); + $worker->registerActivity('orders.simple', static fn (ActivityContext $context): string => 'done'); + self::assertTrue($worker->tick(0)); + self::assertSame([], self::samples($recorder)); + } + + /** @return array */ + public static function unsupportedCapabilities(): array + { + return ['absent' => [null], 'false' => [false], 'integer' => [1], 'text' => ['true'], 'array' => [[]]]; + } + + public function testMissingAgeIsNotZero(): void + { + $recorder = new RecordingMetricsRecorder(); + $diagnostics = []; + [$worker, $client] = $this->worker($recorder, 'activity', [], diagnostics: $diagnostics); + $worker->registerActivity('orders.simple', static fn (ActivityContext $context): string => 'done'); + self::assertTrue($worker->tick(0)); + self::assertSame([], self::samples($recorder)); + self::assertSame([], array_filter($diagnostics, static fn (array $r): bool => $r[0] === 'worker.metrics_invalid_ready_age')); + } + + public function testUndiscoveredCapabilityEmitsNothing(): void + { + $recorder = new RecordingMetricsRecorder(); + [$worker] = $this->worker($recorder, 'activity', ['ready_age_ms' => 1500], discover: false); + $worker->registerActivity('orders.simple', static fn (ActivityContext $context): string => 'done'); + self::assertTrue($worker->tick(0)); + self::assertSame([], self::samples($recorder)); + } + + public function testManagedRegistrationSuppliesCapabilityWithoutExtraDiscovery(): void + { + $recorder = new RecordingMetricsRecorder(); + [$worker, , $transport] = $this->worker($recorder, 'workflow', ['ready_age_ms' => 1500], discover: false); + $worker->registerWorkflow('orders.simple', static function (WorkflowContext $context) use ($worker): string { + $worker->requestShutdown(); + + return 'done'; + }); + $worker->run(0); + self::assertCount(1, self::samples($recorder)); + self::assertSame([], array_filter($transport->requests, static fn (array $r): bool => str_ends_with($r['uri'], '/cluster/info'))); + } + + public function testQueryDispatchHasNoScheduleToStartObservation(): void + { + $recorder = new RecordingMetricsRecorder(); + [$worker] = $this->worker($recorder, 'query', ['ready_age_ms' => 1500, 'query_name' => 'status']); + $called = 0; + $worker->registerQuery('orders.simple', 'status', static function (QueryContext $context) use (&$called): string { + ++$called; + + return 'done'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(1, $called); + self::assertSame([], self::samples($recorder)); + } + + public function testPollTransportRetryStillHasOneDispatchSample(): void + { + $recorder = new RecordingMetricsRecorder(); + $polls = 0; + [$worker, , $transport] = $this->worker($recorder, 'activity', ['ready_age_ms' => 1500], + beforeRequest: static function (string $method, string $uri) use (&$polls): void { + if (str_ends_with($uri, '/activity-tasks/poll') && ++$polls === 1) { + throw new TransportException('connection reset', transientConnectionFailure: true); + } + }); + $called = 0; + $worker->registerActivity('orders.simple', static function (ActivityContext $context) use (&$called): string { + ++$called; + + return 'done'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(1, $called); + self::assertCount(1, self::samples($recorder)); + $requests = array_values(array_filter($transport->requests, static fn (array $r): bool => str_ends_with($r['uri'], '/activity-tasks/poll'))); + self::assertCount(2, $requests); + self::assertSame($requests[0]['body']['poll_request_id'], $requests[1]['body']['poll_request_id']); + } + + public function testCompletionTransportRetryDoesNotRedispatch(): void + { + $recorder = new RecordingMetricsRecorder(); + $completions = 0; + [$worker] = $this->worker($recorder, 'activity', ['ready_age_ms' => 1500], + beforeRequest: static function (string $method, string $uri) use (&$completions): void { + if (str_ends_with($uri, '/complete') && ++$completions === 1) { + throw new TransportException('connection reset', transientConnectionFailure: true); + } + }); + $called = 0; + $worker->registerActivity('orders.simple', static function (ActivityContext $context) use (&$called): string { + ++$called; + + return 'done'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(1, $called); + self::assertSame(2, $completions); + self::assertCount(1, self::samples($recorder)); + } + + public function testActualRedispatchRemainsAProcessObservation(): void + { + $recorder = new RecordingMetricsRecorder(); + [$worker] = $this->worker($recorder, 'activity', ['ready_age_ms' => 1500]); + $worker->registerActivity('orders.simple', static fn (ActivityContext $context): string => 'done'); + self::assertTrue($worker->tick(0)); + self::assertTrue($worker->tick(0)); + $samples = self::samples($recorder); + self::assertCount(2, $samples); + self::assertSame($samples[0], $samples[1], 'A replayed claim retains its age; samples are not durable execution counts.'); + } + + public function testFailingDiagnosticCannotFailTheTask(): void + { + $recorder = new RecordingMetricsRecorder(); + [$worker] = $this->worker($recorder, 'activity', ['ready_age_ms' => -1], throwDiagnostic: true); + $called = 0; + $worker->registerActivity('orders.simple', static function (ActivityContext $context) use (&$called): string { + ++$called; + + return 'done'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(1, $called); + self::assertSame([], self::samples($recorder)); + } + + public function testFailingRecorderCannotFailTheTask(): void + { + $recorder = new class implements MetricsRecorder { + public function increment(string $name, float $value = 1.0, array $tags = []): void + { + throw new \RuntimeException('export unavailable'); + } + public function record(string $name, float $value, array $tags = []): void + { + throw new \RuntimeException('export unavailable'); + } + }; + [$worker] = $this->worker($recorder, 'activity', ['ready_age_ms' => 1500]); + $called = 0; + $worker->registerActivity('orders.simple', static function (ActivityContext $context) use (&$called): string { + ++$called; + + return 'done'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(1, $called); + } + + /** + * @param array $metadata + * @param list}> $diagnostics + * @return array{Worker, Client, FakeTransport} + */ + private function worker(MetricsRecorder $recorder, string $kind, array $metadata, mixed $capability = true, array &$diagnostics = [], bool $discover = true, ?\Closure $beforeRequest = null, bool $throwDiagnostic = false): array + { + $transport = new FakeTransport(handler: static function (string $method, string $uri, array $headers, ?array $body) use ($kind, $metadata, $capability, $beforeRequest): ?array { + $beforeRequest?->__invoke($method, $uri); + if (str_ends_with($uri, '/register')) { + return ['registered' => true, 'server_capabilities' => ['claimed_task_ready_age' => $capability]]; + } + if ($method === 'DELETE') { + return ['deregistered' => true]; + } + if (str_ends_with($uri, '/cluster/info')) { + return ['worker_protocol' => ['server_capabilities' => $capability === null ? [] : ['claimed_task_ready_age' => $capability]]]; + } + if (str_ends_with($uri, '/'.$kind.'-tasks/poll')) { + return ['task' => $metadata + ['task_id' => $kind.'-task', 'workflow_id' => 'private-workflow', + 'run_id' => 'private-run', 'workflow_type' => 'orders.simple', 'workflow_task_attempt' => 1, + 'activity_type' => 'orders.simple', 'activity_attempt_id' => 'private-attempt', + 'lease_owner' => 'worker-1', 'payload_codec' => 'avro', 'history_events' => []], 'poll_status' => 'leased']; + } + if (str_ends_with($uri, '/poll')) { + return ['task' => null, 'poll_status' => 'empty']; + } + if (str_ends_with($uri, '/heartbeat')) { + return ['renewed' => true, 'task_id' => 'workflow-task', 'lease_owner' => $body['lease_owner'], 'workflow_task_attempt' => $body['workflow_task_attempt']]; + } + if (str_ends_with($uri, '/complete')) { + return ['recorded' => true, 'outcome' => 'completed']; + } + throw new \LogicException('Unexpected request '.$method.' '.$uri); + }); + $client = new Client('https://server.example', transport: $transport, metrics: $recorder); + if ($discover) { + $client->workerClusterInfo(); + } + $logger = new class($diagnostics, $throwDiagnostic) extends AbstractLogger { + public function __construct(private array &$events, private bool $throwDiagnostic) {} + public function log($level, $message, array $context = []): void + { + $this->events[] = [(string) $message, $context]; + if ($this->throwDiagnostic && $message === 'worker.metrics_invalid_ready_age') { + throw new \RuntimeException('diagnostic unavailable'); + } + } + }; + $now = 0.0; + $worker = new Worker($client, 'orders', workerId: 'worker-1', + clock: static function () use (&$now): float { return $now; }, + sleeper: static function (int $microseconds) use (&$now): void { $now += $microseconds / 1_000_000; }, + logger: $logger); + + + return [$worker, $client, $transport]; + } + + /** @return list> */ + private static function samples(RecordingMetricsRecorder $recorder): array + { + return array_values(array_filter($recorder->events, static fn (array $event): bool => $event['name'] === self::METRIC)); + } +}