Record capability-gated worker dispatch latency

CHANGELOG.md

@@ -8,6 +8,11 @@ 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.
+
 - `ActivityContext::activityExecutionId` exposes the received logical activity
   execution ID for downstream idempotency keys. Missing or invalid legacy
   metadata stays null; task and attempt IDs are never used as substitutes.

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:

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"]
     }
   ]
 }

src/Client.php

@@ -90,6 +90,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<string, mixed>|null, array<string, mixed>, RequestBudget|null): (?\Closure))|null */
@@ -336,10 +337,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);
     }
@@ -450,12 +453,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<mixed> $input
      * @param array<string, mixed>|null $memo
@@ -1761,7 +1778,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',
@@ -1777,6 +1794,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<string, mixed> */

src/Metrics/MetricsContract.php

@@ -7,18 +7,30 @@
 /** @internal Loads the exact pinned shared registry, not a PHP-specific metric list. */
 final class MetricsContract
 {
-    /** @var array<string, array{counter: string, histogram: string}>|null */
+    /** @var array<string, array{counter?: string, histogram: string}>|null */
     private static ?array $pairs = null;
 
     /** @return array{counter: string, histogram: string} */
     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<string, array{counter: string, histogram: string}> */
+    /** @return array<string, array{counter?: string, histogram: string}> */
     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.');
             }

src/Metrics/MetricsObserver.php

@@ -50,4 +50,20 @@ public function observe(string $scope, float $seconds, array $tags): void
             $this->recording = false;
         }
     }
+
+    /** @param array<string, string> $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;
+        }
+    }
 }

src/Worker.php

@@ -1236,6 +1236,7 @@ private function taskFromPoll(array $response): ?array
     /** @param array<string, mixed> $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<string, mixed> $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<string, mixed> $task */
     private function dispatchPolledTask(string $taskKind, array $task): void
     {

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'));
     }
 

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());
+    }
 }