TD-S097: worker-session activities are unsafe during mixed server rollouts (#117)

Author: durable-workflow-ops

README.md

@@ -44,7 +44,7 @@ curl -X POST http://localhost:8080/api/worker/register \
   -H "Authorization: Bearer $DW_AUTH_TOKEN" \
   -H "Content-Type: application/json" \
   -H "X-Namespace: default" \
-  -H "X-Durable-Workflow-Protocol-Version: 1.0" \
+  -H "X-Durable-Workflow-Protocol-Version: 1.2" \
   -d '{"worker_id":"quickstart-worker","task_queue":"quickstart","runtime":"python"}'
 ```
 
@@ -93,7 +93,7 @@ curl -X POST http://localhost:8080/api/worker/register \
   -H "Authorization: Bearer $DW_AUTH_TOKEN" \
   -H "Content-Type: application/json" \
   -H "X-Namespace: default" \
-  -H "X-Durable-Workflow-Protocol-Version: 1.0" \
+  -H "X-Durable-Workflow-Protocol-Version: 1.2" \
   -d '{"worker_id":"compose-worker","task_queue":"compose","runtime":"python"}'
 ```
 
@@ -333,7 +333,7 @@ curl -X POST $SERVER/api/worker/register \
   -H "Authorization: Bearer $WORKER_TOKEN" \
   -H "Content-Type: application/json" \
   -H "X-Namespace: default" \
-  -H "X-Durable-Workflow-Protocol-Version: 1.0" \
+  -H "X-Durable-Workflow-Protocol-Version: 1.2" \
   -d '{
     "worker_id": "worker-1",
     "task_queue": "order-workers",
@@ -363,7 +363,7 @@ curl -X POST $SERVER/api/worker/workflow-tasks/poll \
   -H "Authorization: Bearer $WORKER_TOKEN" \
   -H "Content-Type: application/json" \
   -H "X-Namespace: default" \
-  -H "X-Durable-Workflow-Protocol-Version: 1.0" \
+  -H "X-Durable-Workflow-Protocol-Version: 1.2" \
   -d '{
     "worker_id": "worker-1",
     "task_queue": "order-workers"
@@ -374,7 +374,7 @@ The response includes the task, its history events, and lease metadata:
 
 ```json
 {
-  "protocol_version": "1.0",
+  "protocol_version": "1.2",
   "task": {
     "task_id": "task-xyz",
     "workflow_id": "order-42",
@@ -400,7 +400,7 @@ curl -X POST $SERVER/api/worker/workflow-tasks/task-xyz/complete \
   -H "Authorization: Bearer $WORKER_TOKEN" \
   -H "Content-Type: application/json" \
   -H "X-Namespace: default" \
-  -H "X-Durable-Workflow-Protocol-Version: 1.0" \
+  -H "X-Durable-Workflow-Protocol-Version: 1.2" \
   -d '{
     "lease_owner": "worker-1",
     "workflow_task_attempt": 1,
@@ -422,7 +422,7 @@ curl -X POST $SERVER/api/worker/workflow-tasks/task-xyz/complete \
   -H "Authorization: Bearer $WORKER_TOKEN" \
   -H "Content-Type: application/json" \
   -H "X-Namespace: default" \
-  -H "X-Durable-Workflow-Protocol-Version: 1.0" \
+  -H "X-Durable-Workflow-Protocol-Version: 1.2" \
   -d '{
     "lease_owner": "worker-1",
     "workflow_task_attempt": 1,
@@ -445,15 +445,15 @@ curl -X POST $SERVER/api/worker/activity-tasks/poll \
   -H "Authorization: Bearer $WORKER_TOKEN" \
   -H "Content-Type: application/json" \
   -H "X-Namespace: default" \
-  -H "X-Durable-Workflow-Protocol-Version: 1.0" \
+  -H "X-Durable-Workflow-Protocol-Version: 1.2" \
   -d '{"worker_id": "worker-1", "task_queue": "order-workers"}'
 
 # Complete (use task_id and activity_attempt_id from the poll response)
 curl -X POST $SERVER/api/worker/activity-tasks/TASK_ID/complete \
   -H "Authorization: Bearer $WORKER_TOKEN" \
   -H "Content-Type: application/json" \
   -H "X-Namespace: default" \
-  -H "X-Durable-Workflow-Protocol-Version: 1.0" \
+  -H "X-Durable-Workflow-Protocol-Version: 1.2" \
   -d '{
     "activity_attempt_id": "ATTEMPT_ID",
     "lease_owner": "worker-1",
@@ -667,8 +667,8 @@ manifests should fail closed.
 - `POST /api/worker/activity-tasks/{id}/fail` — Fail activity task
 - `POST /api/worker/activity-tasks/{id}/heartbeat` — Activity heartbeat
 
-Worker-plane requests must send `X-Durable-Workflow-Protocol-Version: 1.0`, and
-worker-plane responses always echo the same header plus `protocol_version: "1.0"`.
+Worker-plane requests must send `X-Durable-Workflow-Protocol-Version: 1.2`, and
+worker-plane responses always echo the same header plus `protocol_version: "1.2"`.
 Worker requests with bodies follow the same JSON media-type requirement as the
 control plane and return a worker-protocol 415 response for XML, form, or other
 non-JSON body formats.
@@ -811,7 +811,7 @@ future carriers can validate parser behavior without repository-local fixture
 paths. A human-readable summary lives in
 `docs/contracts/external-task-result.md`.
 
-Within worker protocol version `1.0`, `worker_protocol.version`,
+Within worker protocol version `1.2`, `worker_protocol.version`,
 `server_capabilities.long_poll_timeout`, and
 `server_capabilities.supported_workflow_task_commands` are stable contract
 fields. The command-option booleans under `server_capabilities` are additive

app/Http/Controllers/Api/WorkerController.php

@@ -524,6 +524,15 @@ public function completeWorkflowTask(Request $request, string $taskId): JsonResp
 
         $this->validateWorkflowTaskCommandScopes($commands);
 
+        if ($response = $this->guardWorkerSessionCommandsAvailable(
+            $request,
+            $taskId,
+            (int) $validated['workflow_task_attempt'],
+            $commands,
+        )) {
+            return $response;
+        }
+
         if ($response = $this->guardWorkflowTaskOwnership(
             $request,
             $namespace,
@@ -566,6 +575,54 @@ public function completeWorkflowTask(Request $request, string $taskId): JsonResp
         ], $this->workflowOutcomeStatus($outcome['reason']));
     }
 
+    /**
+     * @param  list<array<string, mixed>>  $commands
+     */
+    private function guardWorkerSessionCommandsAvailable(
+        Request $request,
+        string $taskId,
+        int $workflowTaskAttempt,
+        array $commands,
+    ): ?JsonResponse {
+        if (! $this->commandsUseWorkerSessions($commands) || WorkerProtocol::workerSessionsSupported()) {
+            return null;
+        }
+
+        $minimum = WorkerProtocol::workerSessionMinimumProtocolVersion();
+
+        return WorkerProtocol::json([
+            'task_id' => $taskId,
+            'workflow_task_attempt' => $workflowTaskAttempt,
+            'outcome' => 'rejected',
+            'recorded' => false,
+            'reason' => 'worker_sessions_unavailable',
+            'error' => sprintf(
+                'Worker-session activity commands require worker protocol %s or newer.',
+                $minimum,
+            ),
+            'requested_version' => WorkerProtocol::requestVersion($request),
+            'minimum_protocol_version' => $minimum,
+            'remediation' => sprintf(
+                'Complete worker-session workflow tasks through a server node advertising worker protocol %s or newer.',
+                $minimum,
+            ),
+        ], 409);
+    }
+
+    /**
+     * @param  list<array<string, mixed>>  $commands
+     */
+    private function commandsUseWorkerSessions(array $commands): bool
+    {
+        foreach ($commands as $command) {
+            if ($this->hasCommandValue($command, 'worker_session')) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
     /**
      * @param  list<array<string, mixed>>  $commands
      *

app/Http/Controllers/Api/WorkerSessionController.php

@@ -21,6 +21,10 @@ public function create(Request $request): JsonResponse
             return $response;
         }
 
+        if ($response = WorkerProtocol::rejectWorkerSessionsUnavailable($request)) {
+            return $response;
+        }
+
         $namespace = (string) $request->attributes->get('namespace');
 
         $validated = $this->validateSessionRequest($request, workerIdRequired: true);
@@ -59,6 +63,10 @@ public function heartbeat(Request $request, string $sessionId): JsonResponse
             return $response;
         }
 
+        if ($response = WorkerProtocol::rejectWorkerSessionsUnavailable($request)) {
+            return $response;
+        }
+
         $namespace = (string) $request->attributes->get('namespace');
 
         $validated = $request->validate([
@@ -82,6 +90,10 @@ public function close(Request $request, string $sessionId): JsonResponse
             return $response;
         }
 
+        if ($response = WorkerProtocol::rejectWorkerSessionsUnavailable($request)) {
+            return $response;
+        }
+
         $namespace = (string) $request->attributes->get('namespace');
 
         $validated = $request->validate([

app/Support/ActivityTaskPoller.php

@@ -161,6 +161,10 @@ private function claimReadyTask(
                 );
 
                 if ($workerSession !== null) {
+                    if (! WorkerProtocol::workerSessionsSupported()) {
+                        return null;
+                    }
+
                     $admission = $this->workerSessions->admitActivity(
                         $namespace,
                         $worker,

app/Support/WorkerProtocol.php

@@ -68,6 +68,51 @@ public static function rejectUnsupported(Request $request): ?JsonResponse
         ], 400);
     }
 
+    public static function workerSessionMinimumProtocolVersion(): string
+    {
+        $semantics = self::baseWorkerSessionSemantics();
+        $minimum = $semantics['minimum_protocol_version']
+            ?? $semantics['min_worker_protocol_version']
+            ?? ($semantics['rollout_safety']['minimum_protocol_version'] ?? null)
+            ?? self::VERSION;
+
+        return is_string($minimum) && trim($minimum) !== ''
+            ? trim($minimum)
+            : self::VERSION;
+    }
+
+    public static function workerSessionsSupported(): bool
+    {
+        $configured = (string) config('server.worker_protocol.version', self::VERSION);
+
+        return version_compare($configured, self::workerSessionMinimumProtocolVersion(), '>=');
+    }
+
+    public static function rejectWorkerSessionsUnavailable(Request $request): ?JsonResponse
+    {
+        if (self::workerSessionsSupported()) {
+            return null;
+        }
+
+        $configured = (string) config('server.worker_protocol.version', self::VERSION);
+        $minimum = self::workerSessionMinimumProtocolVersion();
+
+        return self::json([
+            'error' => sprintf(
+                'Worker sessions require worker protocol %s or newer.',
+                $minimum,
+            ),
+            'reason' => 'worker_sessions_unavailable',
+            'supported_version' => $configured,
+            'requested_version' => self::requestVersion($request),
+            'minimum_protocol_version' => $minimum,
+            'remediation' => sprintf(
+                'Route worker-session clients only to server nodes advertising worker protocol %s or newer.',
+                $minimum,
+            ),
+        ], 409);
+    }
+
     /**
      * @return list<string>
      */
@@ -108,6 +153,8 @@ public static function supportedWorkflowTaskCommands(): array
      */
     public static function serverCapabilities(): array
     {
+        $workerSessionSupported = self::workerSessionsSupported();
+
         return [
             'long_poll_timeout' => (int) config(
                 'server.polling.timeout',
@@ -128,18 +175,8 @@ public static function serverCapabilities(): array
             'activity_retry_policy' => true,
             'activity_timeouts' => true,
             'local_activities' => self::localActivitySemantics(),
-            'worker_session_verbs' => method_exists(WorkerProtocolVersion::class, 'workerSessionVerbs')
-                ? WorkerProtocolVersion::workerSessionVerbs()
-                : ['create', 'heartbeat', 'close'],
-            'worker_sessions' => method_exists(WorkerProtocolVersion::class, 'workerSessionSemantics')
-                ? WorkerProtocolVersion::workerSessionSemantics()
-                : [
-                    'command_field' => 'worker_session',
-                    'activity_options_field' => 'worker_session',
-                    'lifecycle' => 'lazy_create_on_first_admitted_activity',
-                    'ownership' => 'single_worker_lease_owner',
-                    'verbs' => ['create', 'heartbeat', 'close'],
-                ],
+            'worker_session_verbs' => $workerSessionSupported ? self::workerSessionVerbs() : [],
+            'worker_sessions' => self::workerSessionSemantics($workerSessionSupported),
             'child_workflow_retry_policy' => true,
             'child_workflow_timeouts' => true,
             'parent_close_policy' => true,
@@ -178,6 +215,49 @@ public static function serverCapabilities(): array
         ];
     }
 
+    /**
+     * @return list<string>
+     */
+    private static function workerSessionVerbs(): array
+    {
+        return method_exists(WorkerProtocolVersion::class, 'workerSessionVerbs')
+            ? WorkerProtocolVersion::workerSessionVerbs()
+            : ['create', 'heartbeat', 'close'];
+    }
+
+    /**
+     * @return array<string, mixed>
+     */
+    private static function workerSessionSemantics(bool $supported): array
+    {
+        $minimum = self::workerSessionMinimumProtocolVersion();
+
+        return [
+            ...self::baseWorkerSessionSemantics(),
+            'supported' => $supported,
+            'minimum_protocol_version' => $minimum,
+            ...($supported ? [] : [
+                'unavailable_reason' => 'worker_protocol_version_below_worker_session_minimum',
+            ]),
+        ];
+    }
+
+    /**
+     * @return array<string, mixed>
+     */
+    private static function baseWorkerSessionSemantics(): array
+    {
+        return method_exists(WorkerProtocolVersion::class, 'workerSessionSemantics')
+            ? WorkerProtocolVersion::workerSessionSemantics()
+            : [
+                'command_field' => 'worker_session',
+                'activity_options_field' => 'worker_session',
+                'lifecycle' => 'lazy_create_on_first_admitted_activity',
+                'ownership' => 'single_worker_lease_owner',
+                'verbs' => ['create', 'heartbeat', 'close'],
+            ];
+    }
+
     /**
      * @return array<string, mixed>
      */

composer.lock

@@ -6,6 +6,7 @@
 
 use App\Models\WorkerRegistration;
 use App\Support\NamespaceWorkflowScope;
+use App\Support\WorkerProtocol;
 use Illuminate\Foundation\Testing\RefreshDatabase;
 use Illuminate\Support\Facades\Queue;
 use Tests\Feature\Concerns\ServerTestHelpers;
@@ -284,7 +285,7 @@ private function workerHeaders(string $namespace = 'default'): array
         return [
             'X-Namespace' => $namespace,
             'X-Durable-Workflow-Control-Plane-Version' => '2',
-            'X-Durable-Workflow-Protocol-Version' => '1.0',
+            WorkerProtocol::HEADER => WorkerProtocol::VERSION,
         ];
     }