From 3e10998a6125ca3f4a6a78812d49520d8fc42498 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 05:18:52 +0000 Subject: [PATCH 1/3] Preserve Nexus worker fields and reproduce service dispatch --- app/Http/Controllers/Api/WorkerController.php | 74 +++++++++++++ tests/Feature/WorkerServiceOperationTest.php | 100 ++++++++++++++++++ 2 files changed, 174 insertions(+) create mode 100644 tests/Feature/WorkerServiceOperationTest.php diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php index 973a07f3..35778a8b 100644 --- a/app/Http/Controllers/Api/WorkerController.php +++ b/app/Http/Controllers/Api/WorkerController.php @@ -2,6 +2,8 @@ namespace App\Http\Controllers\Api; +use App\Contracts\AuthProvider; +use App\Http\Middleware\Authenticate; use App\Models\WorkerBuildIdRollout; use App\Models\WorkerRegistration; use App\Support\AvroPayloadEnvelopeResolver; @@ -18,6 +20,7 @@ use App\Support\PayloadCodecContract; use App\Support\PollRequestTaskKindsConflict; use App\Support\QueryTaskQueueUnavailableException; +use App\Support\RouteAuthorizationResource; use App\Support\RuntimeExternalPayloadAudit; use App\Support\RuntimeExternalPayloadException; use App\Support\SearchAttributeValueValidator; @@ -1506,6 +1509,31 @@ public function completeWorkflowTask(Request $request, string $taskId): JsonResp 'commands.*.condition_wait_occurrence_id' => ['nullable', 'string'], 'commands.*.signal_name' => ['nullable', 'string'], 'commands.*.timeout_seconds' => ['nullable', 'integer', 'min:0'], + 'commands.*.endpoint_name' => ['nullable', 'string', 'max:191'], + 'commands.*.service_name' => ['nullable', 'string', 'max:191'], + 'commands.*.operation_name' => ['nullable', 'string', 'max:191'], + 'commands.*.request_payload' => ['nullable'], + 'commands.*.namespace' => ['nullable', 'string', 'max:191'], + 'commands.*.caller_namespace' => ['nullable', 'string', 'max:191'], + 'commands.*.service_call_id' => ['nullable', 'string', 'max:191'], + 'commands.*.idempotency_key' => ['nullable', 'string', 'max:191'], + 'commands.*.mode_override' => ['nullable', 'string', 'in:sync,async'], + 'commands.*.wait_for' => ['nullable', 'string', 'in:accepted,completed'], + 'commands.*.wait_timeout_seconds' => ['nullable', 'integer', 'min:0'], + 'commands.*.target_workflow_instance_id' => ['nullable', 'string', 'max:191'], + 'commands.*.target_workflow_run_id' => ['nullable', 'string', 'max:191'], + 'commands.*.business_key' => ['nullable', 'string', 'max:191'], + 'commands.*.labels' => ['nullable', 'array'], + 'commands.*.memo' => ['nullable', 'array'], + 'commands.*.search_attributes' => ['nullable', 'array'], + 'commands.*.duplicate_start_policy' => ['nullable', 'string'], + 'commands.*.metadata' => ['nullable', 'array'], + 'commands.*.request_payload_reference' => ['nullable', 'string', 'max:191'], + 'commands.*.principal_subject' => ['prohibited'], + 'commands.*.principal_method' => ['prohibited'], + 'commands.*.principal_roles' => ['prohibited'], + 'commands.*.principal_tenant' => ['prohibited'], + 'commands.*.principal_claims' => ['prohibited'], ...WorkflowCommandNormalizer::parallelMetadataValidationRules(), 'commands.*.workflow_stream' => ['nullable', 'array'], 'commands.*.workflow_stream.operation' => ['required_with:commands.*.workflow_stream', 'string', 'in:append,close,error'], @@ -1561,6 +1589,8 @@ public function completeWorkflowTask(Request $request, string $taskId): JsonResp return $response; } + $commands = $this->authorizeServiceOperationCommands($request, (string) $namespace, $commands); + if ($response = $this->guardWorkerSessionCommandsAvailable( $request, $taskId, @@ -2651,6 +2681,49 @@ private function promoteWorkflowFailureExceptionPayload(array $commands): array * @param list> $commands * @return list> */ + private function authorizeServiceOperationCommands(Request $request, string $namespace, array $commands): array + { + foreach ($commands as $index => $command) { + if (($command['type'] ?? null) !== 'start_service_operation') { + continue; + } + + $principal = Authenticate::principal($request); + $targetNamespace = strtolower(trim($command['namespace'] ?? $namespace)); + $callerNamespace = strtolower(trim($command['caller_namespace'] ?? $namespace)); + if ($callerNamespace !== $namespace) { + throw ValidationException::withMessages([ + "commands.{$index}.caller_namespace" => 'The caller namespace must match the leased workflow namespace.', + ]); + } + + $resource = array_replace(app(RouteAuthorizationResource::class)->make($request, ['worker']), [ + 'operation_family' => 'service', + 'operation_name' => 'execute_operation', + 'target_namespace' => $targetNamespace, + 'namespace_name' => $targetNamespace, + 'caller_namespace' => $namespace, + 'service_endpoint_name' => strtolower(trim($command['endpoint_name'])), + 'service_name' => strtolower(trim($command['service_name'])), + 'service_operation_name' => strtolower(trim($command['operation_name'])), + ]); + abort_unless($principal !== null && app(AuthProvider::class)->authorize($principal, 'execute_operation', $resource), 403); + + // A worker owns command intent, not the authenticated caller identity. + $commands[$index] = array_replace($command, [ + 'namespace' => $targetNamespace, + 'caller_namespace' => $namespace, + 'principal_subject' => $principal->subject(), + 'principal_method' => $principal->method(), + 'principal_roles' => $principal->roles(), + 'principal_tenant' => $principal->tenant(), + 'principal_claims' => $principal->claims(), + ]); + } + + return $commands; + } + private function normalizeWorkflowTaskCommandIntegerFields(array $commands): array { $integerFields = [ @@ -2668,6 +2741,7 @@ private function normalizeWorkflowTaskCommandIntegerFields(array $commands): arr 'min_supported', 'max_supported', 'timeout_seconds', + 'wait_timeout_seconds', ]; foreach ($commands as $index => $command) { diff --git a/tests/Feature/WorkerServiceOperationTest.php b/tests/Feature/WorkerServiceOperationTest.php new file mode 100644 index 00000000..53caf195 --- /dev/null +++ b/tests/Feature/WorkerServiceOperationTest.php @@ -0,0 +1,100 @@ +createNamespace('default'); + $this->registerWorker('nexus-worker', 'nexus', supportedWorkflowTypes: ['tests.nexus-caller', 'tests.nexus-target']); + } + + public function test_worker_command_dispatches_a_service_operation_and_persists_its_outcome(): void + { + $this->seedCatalog(); + $task = $this->leaseWorkflow(); + + $this->complete($task, $this->command())->assertOk(); + + $call = WorkflowServiceCall::query()->sole(); + $this->assertSame('started', $call->status->value ?? $call->status); + $this->assertSame('billing', $call->endpoint_name); + $this->assertSame('invoicing', $call->service_name); + $this->assertSame('createinvoice', $call->operation_name); + $this->assertSame($task['run_id'], $call->caller_workflow_run_id); + $this->assertSame('default', $call->caller_namespace); + $target = WorkflowRun::query()->findOrFail($call->linked_workflow_run_id); + $this->assertSame([['invoice' => 42]], Avro::unserialize($target->input)); + $this->assertDatabaseHas('workflow_tasks', ['workflow_run_id' => $target->id, 'status' => 'ready']); + $this->assertDatabaseHas('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted']); + $this->assertSame('completed', WorkflowTask::query()->findOrFail($task['task_id'])->status->value); + } + + private function command(array $options = []): array + { + return $options + [ + 'type' => 'start_service_operation', + 'endpoint_name' => 'billing', + 'service_name' => 'invoicing', + 'operation_name' => 'createinvoice', + 'request_payload' => Avro::envelope([['invoice' => 42]]), + ]; + } + + private function complete(array $task, array $command): \Illuminate\Testing\TestResponse + { + return $this->postJson('/api/worker/workflow-tasks/'.$task['task_id'].'/complete', [ + 'lease_owner' => $task['lease_owner'], + 'workflow_task_attempt' => $task['workflow_task_attempt'], + 'commands' => [$command], + ], $this->workerHeaders()); + } + + private function leaseWorkflow(): array + { + $this->postJson('/api/workflows', [ + 'workflow_id' => 'nexus-caller', 'workflow_type' => 'tests.nexus-caller', + 'task_queue' => 'nexus', 'input' => [], + ], $this->apiHeaders())->assertCreated(); + + return $this->postJson('/api/worker/workflow-tasks/poll', [ + 'worker_id' => 'nexus-worker', 'task_queue' => 'nexus', + ], $this->workerHeaders())->assertOk()->assertJsonPath('poll_status', 'leased')->json('task'); + } + + private function seedCatalog(string $namespace = 'default', array $policy = []): WorkflowServiceOperation + { + $endpoint = WorkflowServiceEndpoint::query()->create(['namespace' => $namespace, 'endpoint_name' => 'billing']); + $service = WorkflowService::query()->create([ + 'workflow_service_endpoint_id' => $endpoint->id, 'namespace' => $namespace, 'service_name' => 'invoicing', + ]); + + return WorkflowServiceOperation::query()->create([ + 'workflow_service_endpoint_id' => $endpoint->id, 'workflow_service_id' => $service->id, + 'namespace' => $namespace, 'operation_name' => 'createinvoice', 'operation_mode' => 'async', + 'handler_binding_kind' => 'start_workflow', 'handler_target_reference' => 'tests.nexus-target', + 'handler_binding' => ['workflow_type' => 'tests.nexus-target'], 'boundary_policy' => $policy, + ]); + } +} From 1621299a58f71c1b60c0e700634a743b3a705e6e Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 05:55:03 +0000 Subject: [PATCH 2/3] Qualify Nexus worker authorization and prepare Server 2.3.1 --- app/Http/Controllers/Api/WorkerController.php | 74 +++++- composer.json | 4 +- composer.lock | 16 +- docker-compose.dedicated-matching.yml | 4 +- docker-compose.memo-rolling.yml | 4 +- docker-compose.published.yml | 4 +- docker-compose.small-cluster.yml | 2 +- docker-compose.yml | 8 +- k8s/README.md | 10 +- k8s/helm/durable-workflow/Chart.yaml | 6 +- k8s/helm/durable-workflow/README.md | 2 +- .../ci/existing-secrets-values.yaml | 2 +- .../ci/ingress-and-hpa-values.yaml | 2 +- .../ci/inline-secrets-values.yaml | 2 +- .../durable-workflow/templates/_helpers.tpl | 2 +- k8s/helm/durable-workflow/values.yaml | 2 +- k8s/helm/examples/values-dev.yaml | 2 +- .../values-external-secrets-operator.yaml | 2 +- .../values-production-existing-secrets.yaml | 2 +- k8s/migration-job.yaml | 2 +- k8s/scheduler-cronjob.yaml | 2 +- k8s/secret.yaml | 2 +- k8s/server-deployment.yaml | 2 +- k8s/worker-deployment.yaml | 2 +- resources/release/source-release.json | 4 +- scripts/k8s-kind-smoke.sh | 2 +- .../WorkerServiceOperationEnvelopeTest.php | 61 +++++ tests/Feature/WorkerServiceOperationTest.php | 246 +++++++++++++++++- .../worker-service-operation-request.json | 21 ++ .../worker-service-operation-request.json | 7 + 30 files changed, 450 insertions(+), 51 deletions(-) create mode 100644 tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php create mode 100644 tests/Fixtures/CodecRegression/worker-service-operation-request.json create mode 100644 tests/Fixtures/CodecRegressionProofs/worker-service-operation-request.json diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php index 35778a8b..a668525f 100644 --- a/app/Http/Controllers/Api/WorkerController.php +++ b/app/Http/Controllers/Api/WorkerController.php @@ -24,6 +24,7 @@ use App\Support\RuntimeExternalPayloadAudit; use App\Support\RuntimeExternalPayloadException; use App\Support\SearchAttributeValueValidator; +use App\Support\ServiceCallBoundary; use App\Support\ServiceModeTimerDispatcher; use App\Support\StreamClosedException; use App\Support\StreamErroredException; @@ -61,6 +62,7 @@ use Workflow\V2\Models\ActivityExecution; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; +use Workflow\V2\Models\WorkflowServiceCall; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Support\StickyExecution; use Workflow\V2\Support\WorkerProtocolVersion; @@ -1526,7 +1528,7 @@ public function completeWorkflowTask(Request $request, string $taskId): JsonResp 'commands.*.labels' => ['nullable', 'array'], 'commands.*.memo' => ['nullable', 'array'], 'commands.*.search_attributes' => ['nullable', 'array'], - 'commands.*.duplicate_start_policy' => ['nullable', 'string'], + 'commands.*.duplicate_start_policy' => ['nullable', 'string', 'in:reject_duplicate,return_existing_active'], 'commands.*.metadata' => ['nullable', 'array'], 'commands.*.request_payload_reference' => ['nullable', 'string', 'max:191'], 'commands.*.principal_subject' => ['prohibited'], @@ -1761,6 +1763,7 @@ function () use ( return $response; } + $this->authorizeServiceOperationReplays($request, (string) $namespace, $taskId, $commands); $commands = $this->canonicalizeWorkflowStreamPayloadCodecs($commands); $commands = app(WorkflowStreamCommandProcessor::class)->process( $taskId, @@ -2709,6 +2712,14 @@ private function authorizeServiceOperationCommands(Request $request, string $nam ]); abort_unless($principal !== null && app(AuthProvider::class)->authorize($principal, 'execute_operation', $resource), 403); + if (isset($command['search_attributes'])) { + $this->searchAttributeValues->validateForNamespace( + $targetNamespace, + $command['search_attributes'], + "commands.{$index}.search_attributes", + ); + } + // A worker owns command intent, not the authenticated caller identity. $commands[$index] = array_replace($command, [ 'namespace' => $targetNamespace, @@ -2724,6 +2735,67 @@ private function authorizeServiceOperationCommands(Request $request, string $nam return $commands; } + /** + * @param list> $commands + */ + private function authorizeServiceOperationReplays(Request $request, string $namespace, string $taskId, array $commands): void + { + foreach ($commands as $index => $command) { + if (($command['type'] ?? null) !== 'start_service_operation') { + continue; + } + + $callId = $command['service_call_id'] ?? null; + $key = $command['idempotency_key'] ?? null; + if ($callId === null && $key === null) { + continue; + } + + $query = WorkflowServiceCall::query()->where('target_namespace', $command['namespace']); + if ($callId !== null) { + $query->whereKey(trim($callId)); + } else { + $query->where('endpoint_name', strtolower(trim($command['endpoint_name']))) + ->where('service_name', strtolower(trim($command['service_name']))) + ->where('operation_name', strtolower(trim($command['operation_name']))) + ->where('idempotency_key', trim($key)) + ->oldest('created_at')->oldest('id'); + } + $call = $query->first(); + if ($call === null && $callId === null) { + continue; + } + + $runId = WorkflowTask::query()->whereKey($taskId)->value('workflow_run_id'); + $principal = Authenticate::principal($request); + abort_unless($call !== null && $principal !== null + && $call->caller_namespace === $namespace + && $call->caller_workflow_run_id === $runId + && $call->endpoint_name === strtolower(trim($command['endpoint_name'])) + && $call->service_name === strtolower(trim($command['service_name'])) + && $call->operation_name === strtolower(trim($command['operation_name'])), 403); + + // Existing call IDs and idempotency keys must not skip current boundary policy. + $operation = $call->operation; + abort_unless($operation !== null && $call->endpoint !== null && $call->service !== null, 403); + $rejection = app(ServiceCallBoundary::class)->replayRejectionFor( + principal: $principal, + callerNamespace: $namespace, + operation: $operation, + endpointName: $call->endpoint_name, + serviceName: $call->service_name, + callerWorkflowInstanceId: $call->caller_workflow_instance_id, + callerWorkflowRunId: $runId, + idempotencyKey: $key, + operationModeOverride: $command['mode_override'] ?? null, + endpointBoundaryPolicy: $call->endpoint->boundary_policy ?? [], + serviceBoundaryPolicy: $call->service->boundary_policy ?? [], + operationBoundaryPolicy: $operation->boundary_policy ?? [], + ); + abort_if($rejection !== null, 403); + } + } + private function normalizeWorkflowTaskCommandIntegerFields(array $commands): array { $integerFields = [ diff --git a/composer.json b/composer.json index 5a74f6cd..15316b2a 100644 --- a/composer.json +++ b/composer.json @@ -6,7 +6,7 @@ "require": { "php": "^8.2", "apache/avro": "^1.12", - "durable-workflow/workflow": "2.0.3", + "durable-workflow/workflow": "2.0.6", "laravel/framework": "^13.0", "laravel/tinker": "^3.0", "league/flysystem-aws-s3-v3": "^3.35.3" @@ -48,7 +48,7 @@ }, "extra": { "durable-workflow": { - "product-train": "2.3.0" + "product-train": "2.3.1" }, "laravel": { "dont-discover": [] diff --git a/composer.lock b/composer.lock index 11388cdd..1f23caea 100644 --- a/composer.lock +++ b/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "5daf3bd5078b60948bf590c25faadf9e", + "content-hash": "bf5c8a06f83c9d62e5a6147dbe6c368e", "packages": [ { "name": "apache/avro", @@ -655,16 +655,16 @@ }, { "name": "durable-workflow/workflow", - "version": "2.0.3", + "version": "2.0.6", "source": { "type": "git", "url": "https://github.com/durable-workflow/workflow.git", - "reference": "9e351efdd3eff38d2305eec248c90392846cec35" + "reference": "af1b743a037e40186f964e357fcea7a0978f91a3" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/9e351efdd3eff38d2305eec248c90392846cec35", - "reference": "9e351efdd3eff38d2305eec248c90392846cec35", + "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/af1b743a037e40186f964e357fcea7a0978f91a3", + "reference": "af1b743a037e40186f964e357fcea7a0978f91a3", "shasum": "" }, "require": { @@ -697,7 +697,7 @@ "dev-main": "2.0.x-dev" }, "durable-workflow": { - "product-train": "2.0.3", + "product-train": "2.0.6", "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json", "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json" } @@ -724,9 +724,9 @@ "description": "Embedded durable workflow runtime and orchestration engine for Laravel applications.", "support": { "issues": "https://github.com/durable-workflow/workflow/issues", - "source": "https://github.com/durable-workflow/workflow/tree/2.0.3" + "source": "https://github.com/durable-workflow/workflow/tree/2.0.6" }, - "time": "2026-09-02T19:15:12+00:00" + "time": "2026-09-08T05:52:28+00:00" }, { "name": "egulias/email-validator", diff --git a/docker-compose.dedicated-matching.yml b/docker-compose.dedicated-matching.yml index 18041965..d3ba33b6 100644 --- a/docker-compose.dedicated-matching.yml +++ b/docker-compose.dedicated-matching.yml @@ -32,13 +32,13 @@ name: durable-workflow-server # daemon reports `shape: dedicated`. # Generated by scripts/ci/sync-source-release.mjs. Do not edit the fallback. -x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.0}} +x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.1}} x-server-environment: &server-environment APP_NAME: "Durable Workflow Server" APP_ENV: ${APP_ENV:-local} DW_SERVER_KEY: ${DW_SERVER_KEY:-} - APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.0}} + APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.1}} APP_DEBUG: ${APP_DEBUG:-false} DB_CONNECTION: mysql DB_HOST: mysql diff --git a/docker-compose.memo-rolling.yml b/docker-compose.memo-rolling.yml index 5404584c..a56d0f50 100644 --- a/docker-compose.memo-rolling.yml +++ b/docker-compose.memo-rolling.yml @@ -49,14 +49,14 @@ services: command: ["server-bootstrap"] environment: <<: *runtime-environment - APP_VERSION: ${APP_VERSION:-2.3.0} + APP_VERSION: ${APP_VERSION:-2.3.1} successor: image: ${DW_MEMO_SUCCESSOR_IMAGE:-durable-workflow/server-memo-rolling:local} ports: !override [] environment: <<: *runtime-environment - APP_VERSION: ${APP_VERSION:-2.3.0} + APP_VERSION: ${APP_VERSION:-2.3.1} DW_SERVER_ID: memo-successor DW_SERVER_TOPOLOGY_SHAPE: standalone_server DW_SERVER_PROCESS_CLASS: server_http_node diff --git a/docker-compose.published.yml b/docker-compose.published.yml index 5f5b3a2d..6f29c801 100644 --- a/docker-compose.published.yml +++ b/docker-compose.published.yml @@ -1,13 +1,13 @@ name: durable-workflow-server # Generated by scripts/ci/sync-source-release.mjs. Do not edit the fallback. -x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.0}} +x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.1}} x-server-environment: &server-environment APP_NAME: "Durable Workflow Server" APP_ENV: ${APP_ENV:-local} DW_SERVER_KEY: ${DW_SERVER_KEY:-} - APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.0}} + APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.1}} APP_DEBUG: ${APP_DEBUG:-false} LOG_CHANNEL: ${LOG_CHANNEL:-stderr} LOG_LEVEL: ${LOG_LEVEL:-info} diff --git a/docker-compose.small-cluster.yml b/docker-compose.small-cluster.yml index 9388da68..4218ea17 100644 --- a/docker-compose.small-cluster.yml +++ b/docker-compose.small-cluster.yml @@ -12,7 +12,7 @@ x-server-build: &server-build x-server-environment: &server-environment APP_NAME: "Durable Workflow Server" APP_ENV: testing - APP_VERSION: ${APP_VERSION:-2.3.0} + APP_VERSION: ${APP_VERSION:-2.3.1} APP_DEBUG: "false" DW_SERVER_KEY: ${DW_SERVER_KEY:-base64:5Zt4nUhlCm3DD0nLXZJQdHiwPfb56yGo9gNV/g3jYbY=} DB_CONNECTION: ${DW_SMALL_CLUSTER_DB:-mysql} diff --git a/docker-compose.yml b/docker-compose.yml index e7465a75..e929b780 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -15,7 +15,7 @@ services: DW_SERVER_KEY: "${DW_SERVER_KEY:-}" DW_SERVER_TOPOLOGY_SHAPE: standalone_server DW_SERVER_PROCESS_CLASS: server_http_node - APP_VERSION: "${APP_VERSION:-2.3.0}" + APP_VERSION: "${APP_VERSION:-2.3.1}" APP_DEBUG: "false" DB_CONNECTION: mysql DB_HOST: mysql @@ -62,7 +62,7 @@ services: APP_NAME: "Durable Workflow Server" APP_ENV: local DW_SERVER_KEY: "${DW_SERVER_KEY:-}" - APP_VERSION: "${APP_VERSION:-2.3.0}" + APP_VERSION: "${APP_VERSION:-2.3.1}" APP_DEBUG: "false" DB_CONNECTION: mysql DB_HOST: mysql @@ -123,7 +123,7 @@ services: DW_SERVER_KEY: "${DW_SERVER_KEY:-}" DW_SERVER_TOPOLOGY_SHAPE: standalone_server DW_SERVER_PROCESS_CLASS: worker_node - APP_VERSION: "${APP_VERSION:-2.3.0}" + APP_VERSION: "${APP_VERSION:-2.3.1}" DB_CONNECTION: mysql DB_HOST: mysql DB_PORT: 3306 @@ -176,7 +176,7 @@ services: DW_SERVER_KEY: "${DW_SERVER_KEY:-}" DW_SERVER_TOPOLOGY_SHAPE: standalone_server DW_SERVER_PROCESS_CLASS: scheduler_node - APP_VERSION: "${APP_VERSION:-2.3.0}" + APP_VERSION: "${APP_VERSION:-2.3.1}" DB_CONNECTION: mysql DB_HOST: mysql DB_PORT: 3306 diff --git a/k8s/README.md b/k8s/README.md index d2547bf6..f3030d91 100644 --- a/k8s/README.md +++ b/k8s/README.md @@ -13,7 +13,7 @@ The checked-in manifests are synchronized with the repository's stable source release and pin its Docker Hub tag: ```text -durableworkflow/server:2.3.0 +durableworkflow/server:2.3.1 ``` Before production use, patch every workload image to the exact published tag or @@ -21,15 +21,15 @@ digest you intend to run: ```bash kubectl set image -n durable-workflow deploy/durable-workflow-server \ - server=durableworkflow/server:2.3.0 + server=durableworkflow/server:2.3.1 kubectl set image -n durable-workflow deploy/durable-workflow-worker \ - worker=durableworkflow/server:2.3.0 + worker=durableworkflow/server:2.3.1 kubectl set image -n durable-workflow cronjob/durable-workflow-scheduler \ - scheduler=durableworkflow/server:2.3.0 + scheduler=durableworkflow/server:2.3.1 ``` GitHub Container Registry publishes the same release line at -`ghcr.io/durable-workflow/server:2.3.0`. Digest pinning is preferred for strict +`ghcr.io/durable-workflow/server:2.3.1`. Digest pinning is preferred for strict change control. The manifests expect you to provide: diff --git a/k8s/helm/durable-workflow/Chart.yaml b/k8s/helm/durable-workflow/Chart.yaml index 316ce907..b9905d06 100644 --- a/k8s/helm/durable-workflow/Chart.yaml +++ b/k8s/helm/durable-workflow/Chart.yaml @@ -5,11 +5,11 @@ type: application # The chart's own semver version. Bumped on every chart release; treated as # independent of the server image version (appVersion). Breaking-change rules # for this version live in docs/helm-upgrading.md alongside the chart. -version: 0.1.79 +version: 0.1.80 # The immutable Durable Workflow Server identity this chart release packages. # The onboarding default in values.yaml and appVersion are generated from the # checked-in source release record. -appVersion: "2.3.0" +appVersion: "2.3.1" kubeVersion: ">=1.27.0-0" home: https://durable-workflow.github.io/docs/2.0/deployment sources: @@ -30,7 +30,7 @@ annotations: # exact commit that most recently changed the packaged chart. org.opencontainers.image.source: https://github.com/durable-workflow/server dev.durable-workflow.source-revision: "unreleased" - dev.durable-workflow.image-reference: "docker.io/durableworkflow/server:2.3.0" + dev.durable-workflow.image-reference: "docker.io/durableworkflow/server:2.3.1" artifacthub.io/license: MIT artifacthub.io/category: integration-delivery # Free-form changelog for the current chart release shown by Artifact Hub. diff --git a/k8s/helm/durable-workflow/README.md b/k8s/helm/durable-workflow/README.md index b95e46c2..40bac76e 100644 --- a/k8s/helm/durable-workflow/README.md +++ b/k8s/helm/durable-workflow/README.md @@ -63,7 +63,7 @@ helm install durable-workflow ./k8s/helm/durable-workflow \ ```yaml image: - tag: "2.3.0" + tag: "2.3.1" # Pin a digest in production: # digest: "sha256:abc123..." # memoPayloadStorage: "raw-json-v1" # Required for a digest or custom image. diff --git a/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml b/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml index 66d9ac9c..c528f9f8 100644 --- a/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml +++ b/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml @@ -1,7 +1,7 @@ # CI fixture: GitOps / externally-managed-secret path. The chart consumes # existing Secrets and renders no Secret resources of its own. image: - tag: "2.3.0" + tag: "2.3.1" externalDatabase: connection: pgsql diff --git a/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml b/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml index 73b22c62..c1102f22 100644 --- a/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml +++ b/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml @@ -1,6 +1,6 @@ # CI fixture: ingress + autoscaling enabled. Exercises optional templates. image: - tag: "2.3.0" + tag: "2.3.1" externalDatabase: connection: mysql diff --git a/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml b/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml index b6e89e42..020622aa 100644 --- a/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml +++ b/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml @@ -2,7 +2,7 @@ # chart's render path is exercised end-to-end. Real deployments should use # existingSecret instead. image: - tag: "2.3.0" + tag: "2.3.1" externalDatabase: connection: mysql diff --git a/k8s/helm/durable-workflow/templates/_helpers.tpl b/k8s/helm/durable-workflow/templates/_helpers.tpl index d5b3f32d..2f99df6b 100644 --- a/k8s/helm/durable-workflow/templates/_helpers.tpl +++ b/k8s/helm/durable-workflow/templates/_helpers.tpl @@ -88,7 +88,7 @@ resolved by an explicit capability declaration or an existing workload marker. {{- define "durable-workflow.memoPayloadStorageForImage" -}} {{- $image := toString . -}} {{- $normalized := regexReplaceAll "^index\\.docker\\.io/" $image "docker.io/" -}} -{{- if eq $normalized "docker.io/durableworkflow/server:2.3.0" -}} +{{- if eq $normalized "docker.io/durableworkflow/server:2.3.1" -}} dual-v1 {{- else if regexMatch "^docker\\.io/durableworkflow/server:2\\.0\\.0-rc\\.[0-9]+$" $normalized -}} {{- $releaseCandidate := atoi (regexFind "[0-9]+$" $normalized) -}} diff --git a/k8s/helm/durable-workflow/values.yaml b/k8s/helm/durable-workflow/values.yaml index a78532db..40d8fcf6 100644 --- a/k8s/helm/durable-workflow/values.yaml +++ b/k8s/helm/durable-workflow/values.yaml @@ -21,7 +21,7 @@ image: registry: docker.io repository: durableworkflow/server # Generated by scripts/ci/sync-source-release.mjs. Do not edit this default. - tag: "2.3.0" + tag: "2.3.1" # Optional digest pin. When set, takes precedence over tag for change control. # Example: "sha256:abc123..." digest: "" diff --git a/k8s/helm/examples/values-dev.yaml b/k8s/helm/examples/values-dev.yaml index 1d0076ec..98f30d6b 100644 --- a/k8s/helm/examples/values-dev.yaml +++ b/k8s/helm/examples/values-dev.yaml @@ -3,7 +3,7 @@ # shape in production. image: - tag: "2.3.0" + tag: "2.3.1" externalDatabase: connection: mysql diff --git a/k8s/helm/examples/values-external-secrets-operator.yaml b/k8s/helm/examples/values-external-secrets-operator.yaml index e65e2d42..da7ebeb1 100644 --- a/k8s/helm/examples/values-external-secrets-operator.yaml +++ b/k8s/helm/examples/values-external-secrets-operator.yaml @@ -5,7 +5,7 @@ # concern. image: - tag: "2.3.0" + tag: "2.3.1" externalDatabase: connection: pgsql diff --git a/k8s/helm/examples/values-production-existing-secrets.yaml b/k8s/helm/examples/values-production-existing-secrets.yaml index 9c713292..cfd22cbb 100644 --- a/k8s/helm/examples/values-production-existing-secrets.yaml +++ b/k8s/helm/examples/values-production-existing-secrets.yaml @@ -10,7 +10,7 @@ image: repository: durable-workflow/server # Pin a digest in production for change-control auditability. digest: "" # e.g. "sha256:abc123..." - tag: "2.3.0" + tag: "2.3.1" externalDatabase: connection: pgsql diff --git a/k8s/migration-job.yaml b/k8s/migration-job.yaml index 2327df49..54cd27c0 100644 --- a/k8s/migration-job.yaml +++ b/k8s/migration-job.yaml @@ -13,7 +13,7 @@ spec: restartPolicy: OnFailure containers: - name: migrate - image: durableworkflow/server:2.3.0 + image: durableworkflow/server:2.3.1 command: ["server-entrypoint"] args: ["server-bootstrap"] envFrom: diff --git a/k8s/scheduler-cronjob.yaml b/k8s/scheduler-cronjob.yaml index 1c201f4f..b93eece3 100644 --- a/k8s/scheduler-cronjob.yaml +++ b/k8s/scheduler-cronjob.yaml @@ -24,7 +24,7 @@ spec: restartPolicy: Never containers: - name: scheduler - image: durableworkflow/server:2.3.0 + image: durableworkflow/server:2.3.1 command: ["server-entrypoint"] args: ["sh", "-c", "php artisan schedule:evaluate --limit=100 --json; php artisan activity:timeout-enforce --limit=100; if php artisan list --raw | grep -q '^external-payloads:cleanup '; then php artisan external-payloads:cleanup --limit=100 --json; fi; php artisan history:prune --limit=100"] envFrom: diff --git a/k8s/secret.yaml b/k8s/secret.yaml index c9189209..94895a3e 100644 --- a/k8s/secret.yaml +++ b/k8s/secret.yaml @@ -12,7 +12,7 @@ metadata: app.kubernetes.io/name: durable-workflow data: APP_NAME: "Durable Workflow Server" - APP_VERSION: "2.3.0" + APP_VERSION: "2.3.1" APP_ENV: production APP_DEBUG: "false" DB_CONNECTION: mysql diff --git a/k8s/server-deployment.yaml b/k8s/server-deployment.yaml index 030255ba..6da6218e 100644 --- a/k8s/server-deployment.yaml +++ b/k8s/server-deployment.yaml @@ -23,7 +23,7 @@ spec: spec: containers: - name: server - image: durableworkflow/server:2.3.0 + image: durableworkflow/server:2.3.1 ports: - containerPort: 8080 name: http diff --git a/k8s/worker-deployment.yaml b/k8s/worker-deployment.yaml index 476a8b42..9edb377d 100644 --- a/k8s/worker-deployment.yaml +++ b/k8s/worker-deployment.yaml @@ -19,7 +19,7 @@ spec: spec: containers: - name: worker - image: durableworkflow/server:2.3.0 + image: durableworkflow/server:2.3.1 command: ["server-entrypoint"] args: ["php", "artisan", "queue:work", "--sleep=1", "--tries=3", "--max-time=3600"] envFrom: diff --git a/resources/release/source-release.json b/resources/release/source-release.json index cd4785c8..797cd269 100644 --- a/resources/release/source-release.json +++ b/resources/release/source-release.json @@ -1,9 +1,9 @@ { "schema": "durable-workflow.server.source-release/v1", "server": { - "version": "2.3.0" + "version": "2.3.1" }, "helm_chart": { - "version": "0.1.79" + "version": "0.1.80" } } diff --git a/scripts/k8s-kind-smoke.sh b/scripts/k8s-kind-smoke.sh index 832b7783..21da3f4b 100755 --- a/scripts/k8s-kind-smoke.sh +++ b/scripts/k8s-kind-smoke.sh @@ -7,7 +7,7 @@ cluster="${K8S_SMOKE_CLUSTER:-durable-workflow-server-smoke}" image="${K8S_SMOKE_IMAGE:-durableworkflow/server:k8s-smoke}" # Generated by scripts/ci/sync-source-release.mjs so the smoke replaces the # same default shipped by the public manifests. -manifest_image="durableworkflow/server:2.3.0" +manifest_image="durableworkflow/server:2.3.1" kind_node_image="${K8S_SMOKE_KIND_NODE_IMAGE:-kindest/node:v1.29.4}" artifact_dir="${K8S_SMOKE_ARTIFACT_DIR:-/tmp/durable-workflow-k8s-kind-smoke-artifacts}" rendered_dir="${artifact_dir}/rendered-manifests" diff --git a/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php b/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php new file mode 100644 index 00000000..31c7fc79 --- /dev/null +++ b/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php @@ -0,0 +1,61 @@ +createNamespace('default'); + $this->registerWorker('nexus-worker', 'nexus'); + $endpoint = WorkflowServiceEndpoint::query()->create(['namespace' => 'default', 'endpoint_name' => 'billing']); + $service = WorkflowService::query()->create([ + 'namespace' => 'default', 'workflow_service_endpoint_id' => $endpoint->id, 'service_name' => 'invoicing', + ]); + WorkflowServiceOperation::query()->create([ + 'namespace' => 'default', 'workflow_service_endpoint_id' => $endpoint->id, + 'workflow_service_id' => $service->id, 'operation_name' => 'createinvoice', + 'operation_mode' => 'async', 'handler_binding_kind' => 'start_workflow', + 'handler_target_reference' => 'tests.invoice', 'handler_binding' => ['workflow_type' => 'tests.invoice'], + ]); + $this->postJson('/api/workflows', [ + 'workflow_id' => 'nexus-caller', 'workflow_type' => 'tests.external-greeting-workflow', 'task_queue' => 'nexus', + ], $this->apiHeaders())->assertCreated(); + $task = $this->postJson('/api/worker/workflow-tasks/poll', [ + 'worker_id' => 'nexus-worker', 'task_queue' => 'nexus', + ], $this->workerHeaders())->assertOk()->assertJsonPath('poll_status', 'leased')->json('task'); + + ServerCodecRegressionFixtureExecutor::exercise(function () use ($task): void { + $this->postJson('/api/worker/workflow-tasks/'.$task['task_id'].'/complete', [ + 'lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt'], + 'commands' => [[ + 'type' => 'start_service_operation', 'endpoint_name' => 'billing', 'service_name' => 'invoicing', + 'operation_name' => 'createinvoice', + 'request_payload' => ['codec' => 'avro', 'blob' => 'wwHioz3/VYAiNwwCDgIOaW52b2ljZQRUAAA='], + ]], + ], $this->workerHeaders())->assertOk(); + $call = WorkflowServiceCall::query()->sole(); + $this->assertSame('started', $call->status); + $this->assertNotNull($call->linked_workflow_run_id); + $this->assertDatabaseHas('workflow_history_events', [ + 'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted', + ]); + }); + } +} diff --git a/tests/Feature/WorkerServiceOperationTest.php b/tests/Feature/WorkerServiceOperationTest.php index 53caf195..f5d7d5a8 100644 --- a/tests/Feature/WorkerServiceOperationTest.php +++ b/tests/Feature/WorkerServiceOperationTest.php @@ -6,9 +6,12 @@ use Illuminate\Foundation\Testing\RefreshDatabase; use Illuminate\Support\Facades\Queue; +use Illuminate\Testing\TestResponse; +use PHPUnit\Framework\Attributes\DataProvider; use Tests\Feature\Concerns\ServerTestHelpers; use Tests\TestCase; use Workflow\Serializers\Avro; +use Workflow\V2\Contracts\ServiceControlPlane; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowService; @@ -34,21 +37,251 @@ public function test_worker_command_dispatches_a_service_operation_and_persists_ { $this->seedCatalog(); $task = $this->leaseWorkflow(); + $this->authenticateWorker(); $this->complete($task, $this->command())->assertOk(); $call = WorkflowServiceCall::query()->sole(); - $this->assertSame('started', $call->status->value ?? $call->status); + $this->assertSame('started', $call->status); $this->assertSame('billing', $call->endpoint_name); $this->assertSame('invoicing', $call->service_name); $this->assertSame('createinvoice', $call->operation_name); $this->assertSame($task['run_id'], $call->caller_workflow_run_id); $this->assertSame('default', $call->caller_namespace); + $this->assertSame('nexus-test-principal', $call->caller_principal_subject); + $this->assertSame(['worker'], $call->caller_principal_roles); + $this->assertSame('default', $call->caller_principal_tenant); $target = WorkflowRun::query()->findOrFail($call->linked_workflow_run_id); - $this->assertSame([['invoice' => 42]], Avro::unserialize($target->input)); + $this->assertSame([['invoice' => 42]], Avro::unserialize($target->arguments)); $this->assertDatabaseHas('workflow_tasks', ['workflow_run_id' => $target->id, 'status' => 'ready']); $this->assertDatabaseHas('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted']); $this->assertSame('completed', WorkflowTask::query()->findOrFail($task['task_id'])->status->value); + + $this->registerWorker('invoice-worker', 'invoices', supportedWorkflowTypes: ['tests.nexus-target']); + $targetTask = $this->postJson('/api/worker/workflow-tasks/poll', [ + 'worker_id' => 'invoice-worker', 'task_queue' => 'invoices', + ], $this->workerHeaders())->assertOk()->assertJsonPath('poll_status', 'leased')->json('task'); + $this->assertSame($target->id, $targetTask['run_id']); + $this->complete($targetTask, [ + 'type' => 'complete_workflow', 'result' => Avro::envelope(['invoice' => 42, 'accepted' => true]), + ])->assertOk(); + $this->assertSame('completed', $target->fresh()->status->value); + $this->assertSame(['invoice' => 42, 'accepted' => true], Avro::unserialize($target->fresh()->output)); + } + + public function test_supported_options_reach_the_service_control_plane(): void + { + $task = $this->leaseWorkflow(); + $this->authenticateWorker(); + $options = [ + 'namespace' => 'default', 'caller_namespace' => 'default', 'idempotency_key' => 'invoice-42', + 'mode_override' => 'async', 'wait_for' => 'accepted', 'wait_timeout_seconds' => '0', + 'target_workflow_instance_id' => 'invoice-42', 'target_workflow_run_id' => 'target-run', + 'connection' => 'database', 'queue' => 'invoices', 'business_key' => 'business-42', + 'labels' => ['department' => 'finance'], 'memo' => ['display' => 'Invoice 42'], + 'search_attributes' => [], 'duplicate_start_policy' => 'return_existing_active', + 'metadata' => ['trace' => 'trace-42'], 'request_payload_reference' => 'request-42', + ]; + $this->mock(ServiceControlPlane::class)->shouldReceive('execute')->once() + ->withArgs(function (string $endpoint, string $service, string $operation, array $actual) use ($options, $task): bool { + $this->assertSame(['billing', 'invoicing', 'createinvoice'], [$endpoint, $service, $operation]); + foreach ($options as $name => $value) { + if ($name === 'metadata') { + $this->assertSame($value['trace'], $actual['metadata']['trace']); + } else { + $this->assertSame($name === 'wait_timeout_seconds' ? 0 : $value, $actual[$name], $name); + } + } + $this->assertSame([['invoice' => 42]], $actual['arguments']); + $this->assertSame('avro', $actual['payload_codec']); + $this->assertSame('nexus-test-principal', $actual['principal_subject']); + $this->assertSame(['worker'], $actual['principal_roles']); + $this->assertSame($task['run_id'], $actual['caller_workflow_run_id']); + + return true; + })->andReturn(['accepted' => true, 'service_call_id' => 'call-42', 'status' => 'started']); + + $this->complete($task, $this->command($options))->assertOk(); + } + + #[DataProvider('invalidOptions')] + public function test_invalid_operation_commands_do_not_commit_history_or_consume_the_lease(array $options, string $field): void + { + $task = $this->leaseWorkflow(); + $this->authenticateWorker(); + $history = WorkflowHistoryEvent::query()->count(); + + $this->complete($task, $this->command($options))->assertUnprocessable()->assertJsonValidationErrors('commands.0.'.$field); + + $this->assertSame($history, WorkflowHistoryEvent::query()->count()); + $this->assertSame('leased', WorkflowTask::query()->findOrFail($task['task_id'])->status->value); + $this->assertDatabaseCount('workflow_service_calls', 0); + } + + public static function invalidOptions(): array + { + $cases = [ + [['endpoint_name' => ''], 'endpoint_name'], + [['service_name' => []], 'service_name'], + [['operation_name' => 42], 'operation_name'], + [['mode_override' => 'parallel'], 'mode_override'], + [['wait_for' => 'forever'], 'wait_for'], + [['wait_timeout_seconds' => -1], 'wait_timeout_seconds'], + [['labels' => 'not-an-array'], 'labels'], + [['memo' => 'not-an-array'], 'memo'], + [['metadata' => 'not-an-array'], 'metadata'], + [['request_payload' => 'not-avro'], 'request_payload'], + [['request_payload' => ['codec' => 'avro', 'blob' => 'not-avro']], 'request_payload.blob'], + [['duplicate_start_policy' => 'replace_anything'], 'duplicate_start_policy'], + [['caller_namespace' => 'other-tenant'], 'caller_namespace'], + ]; + foreach (['principal_subject' => 'admin', 'principal_method' => 'system', 'principal_roles' => ['admin'], + 'principal_tenant' => 'other-tenant', 'principal_claims' => ['admin' => true]] as $field => $value) { + $cases[] = [[$field => $value], $field]; + } + + return $cases; + } + + public function test_a_namespace_scoped_worker_cannot_target_a_different_namespace(): void + { + $this->createNamespace('other-tenant'); + $this->seedCatalog('other-tenant'); + $task = $this->leaseWorkflow(); + $this->authenticateWorker(); + $history = WorkflowHistoryEvent::query()->count(); + + $this->complete($task, $this->command(['namespace' => 'other-tenant']))->assertForbidden(); + $this->assertSame($history, WorkflowHistoryEvent::query()->count()); + $this->assertDatabaseCount('workflow_service_calls', 0); + } + + public function test_an_unscoped_worker_can_use_an_authorized_cross_namespace_operation(): void + { + $this->createNamespace('other-tenant'); + $this->seedCatalog('other-tenant', ['caller_namespaces' => ['allow' => ['default']], 'required_roles' => ['worker']]); + $task = $this->leaseWorkflow(); + $this->authenticateWorker(null); + + $this->complete($task, $this->command(['namespace' => 'other-tenant']))->assertOk(); + $call = WorkflowServiceCall::query()->sole(); + $this->assertSame('other-tenant', $call->target_namespace); + $this->assertSame('default', $call->caller_namespace); + $this->assertSame('started', $call->status); + $this->assertSame('other-tenant', WorkflowRun::query()->findOrFail($call->linked_workflow_run_id)->namespace); + } + + public function test_catalog_role_policy_receives_the_authenticated_worker_identity(): void + { + $this->seedCatalog(policy: ['required_roles' => ['admin']]); + $task = $this->leaseWorkflow(); + $this->authenticateWorker(); + + $this->complete($task, $this->command())->assertOk(); + $call = WorkflowServiceCall::query()->sole(); + $this->assertSame('rejected_forbidden', $call->outcome->value); + $this->assertNull($call->linked_workflow_run_id); + $this->assertDatabaseHas('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallFailed']); + } + + public function test_stale_task_attempt_cannot_dispatch_a_service_operation(): void + { + $this->seedCatalog(); + $task = $this->leaseWorkflow(); + $this->authenticateWorker(); + $task['workflow_task_attempt']++; + + $this->complete($task, $this->command())->assertConflict(); + $this->assertDatabaseCount('workflow_service_calls', 0); + } + + #[DataProvider('replaySelectors')] + public function test_reusing_an_owned_service_call_does_not_dispatch_a_second_target(string $selector): void + { + $this->seedCatalog(); + $task = $this->leaseWorkflow(); + $this->authenticateWorker(); + $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk(); + $call = WorkflowServiceCall::query()->sole(); + $task = $this->pollWorkflow(); + + $this->complete($task, $this->command([$selector => $selector === 'service_call_id' ? $call->id : 'invoice-42']))->assertOk(); + + $this->assertDatabaseCount('workflow_service_calls', 1); + $this->assertDatabaseCount('workflow_runs', 2); + } + + #[DataProvider('replaySelectors')] + public function test_service_call_replays_cannot_reuse_another_caller_namespace(string $selector): void + { + $this->seedCatalog(); + $task = $this->leaseWorkflow(); + $this->authenticateWorker(); + $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk(); + $call = WorkflowServiceCall::query()->sole(); + $call->update(['caller_namespace' => 'other-tenant']); + $task = $this->pollWorkflow(); + $history = WorkflowHistoryEvent::query()->count(); + + $this->complete($task, $this->command([$selector => $selector === 'service_call_id' ? $call->id : 'invoice-42']))->assertForbidden(); + + $this->assertSame($history, WorkflowHistoryEvent::query()->count()); + $this->assertSame('leased', WorkflowTask::query()->findOrFail($task['task_id'])->status->value); + $this->assertDatabaseCount('workflow_service_calls', 1); + } + + #[DataProvider('replaySelectors')] + public function test_service_call_replays_are_authorized_against_current_catalog_policy(string $selector): void + { + $operation = $this->seedCatalog(); + $task = $this->leaseWorkflow(); + $this->authenticateWorker(); + $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk(); + $call = WorkflowServiceCall::query()->sole(); + $operation->update(['boundary_policy' => ['required_roles' => ['admin']]]); + $task = $this->pollWorkflow(); + $history = WorkflowHistoryEvent::query()->count(); + + $this->complete($task, $this->command([$selector => $selector === 'service_call_id' ? $call->id : 'invoice-42']))->assertForbidden(); + + $this->assertSame($history, WorkflowHistoryEvent::query()->count()); + $this->assertDatabaseCount('workflow_runs', 2); + } + + public static function replaySelectors(): array + { + return [['service_call_id'], ['idempotency_key']]; + } + + #[DataProvider('replaySelectors')] + public function test_service_call_replays_cannot_reuse_another_workflow_run(string $selector): void + { + $this->seedCatalog(); + $task = $this->leaseWorkflow(); + $this->authenticateWorker(); + $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk(); + $call = WorkflowServiceCall::query()->sole(); + $call->update(['caller_workflow_run_id' => $call->linked_workflow_run_id]); + $task = $this->pollWorkflow(); + $history = WorkflowHistoryEvent::query()->count(); + + $this->complete($task, $this->command([$selector => $selector === 'service_call_id' ? $call->id : 'invoice-42']))->assertForbidden(); + + $this->assertSame($history, WorkflowHistoryEvent::query()->count()); + $this->assertSame('leased', WorkflowTask::query()->findOrFail($task['task_id'])->status->value); + $this->assertDatabaseCount('workflow_service_calls', 1); + } + + private function authenticateWorker(?string $tenant = 'default'): void + { + config([ + 'server.auth.driver' => 'token', 'server.auth.backward_compatible' => false, + 'server.auth.principal_tokens' => json_encode([[ + 'token' => 'nexus-test-token', 'subject' => 'nexus-test-principal', + 'roles' => ['worker'], 'tenant' => $tenant, + ]]), + ]); + $this->withHeader('Authorization', 'Bearer nexus-test-token'); } private function command(array $options = []): array @@ -62,7 +295,7 @@ private function command(array $options = []): array ]; } - private function complete(array $task, array $command): \Illuminate\Testing\TestResponse + private function complete(array $task, array $command): TestResponse { return $this->postJson('/api/worker/workflow-tasks/'.$task['task_id'].'/complete', [ 'lease_owner' => $task['lease_owner'], @@ -78,6 +311,11 @@ private function leaseWorkflow(): array 'task_queue' => 'nexus', 'input' => [], ], $this->apiHeaders())->assertCreated(); + return $this->pollWorkflow(); + } + + private function pollWorkflow(): array + { return $this->postJson('/api/worker/workflow-tasks/poll', [ 'worker_id' => 'nexus-worker', 'task_queue' => 'nexus', ], $this->workerHeaders())->assertOk()->assertJsonPath('poll_status', 'leased')->json('task'); @@ -94,7 +332,7 @@ private function seedCatalog(string $namespace = 'default', array $policy = []): 'workflow_service_endpoint_id' => $endpoint->id, 'workflow_service_id' => $service->id, 'namespace' => $namespace, 'operation_name' => 'createinvoice', 'operation_mode' => 'async', 'handler_binding_kind' => 'start_workflow', 'handler_target_reference' => 'tests.nexus-target', - 'handler_binding' => ['workflow_type' => 'tests.nexus-target'], 'boundary_policy' => $policy, + 'handler_binding' => ['workflow_type' => 'tests.nexus-target', 'queue' => 'invoices'], 'boundary_policy' => $policy, ]); } } diff --git a/tests/Fixtures/CodecRegression/worker-service-operation-request.json b/tests/Fixtures/CodecRegression/worker-service-operation-request.json new file mode 100644 index 00000000..79bd591f --- /dev/null +++ b/tests/Fixtures/CodecRegression/worker-service-operation-request.json @@ -0,0 +1,21 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.codec-regression/v1", + "id": "worker-service-operation-request", + "protocol": { + "codec": "avro", + "schema": "durable_workflow.protocol.Value", + "version": "1", + "fingerprint": "e2a33dff55802237" + }, + "bindings": ["php", "python", "rust"], + "value": { + "type": "array", + "items": [{"type": "map", "entries": [{"key": "invoice", "value": {"type": "long", "value": "42"}}]}] + }, + "framing": { + "encoding": "avro-single-object", + "wire_base64": "wwHioz3/VYAiNwwCDgIOaW52b2ljZQRUAAA=" + }, + "failure_policy": {"operation": "round_trip", "error": null} +} diff --git a/tests/Fixtures/CodecRegressionProofs/worker-service-operation-request.json b/tests/Fixtures/CodecRegressionProofs/worker-service-operation-request.json new file mode 100644 index 00000000..09bf0004 --- /dev/null +++ b/tests/Fixtures/CodecRegressionProofs/worker-service-operation-request.json @@ -0,0 +1,7 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/server-codec-counterfactual-schema.json", + "proof_schema": "durable-workflow.server-codec-counterfactual/v1", + "fixture": "tests/Fixtures/CodecRegression/worker-service-operation-request.json", + "test": "tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php", + "boundaries": ["app/Http/Controllers/Api/WorkerController.php"] +} From cb75a218ce7529e873909656863f8772214be117 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 06:01:07 +0000 Subject: [PATCH 3/3] Use minimal portable Nexus request in regression fixture --- .../CodecRegression/WorkerServiceOperationEnvelopeTest.php | 2 +- .../CodecRegression/worker-service-operation-request.json | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php b/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php index 31c7fc79..d4d7a319 100644 --- a/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php +++ b/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php @@ -47,7 +47,7 @@ public function test_service_operation_request_survives_worker_validation(): voi 'commands' => [[ 'type' => 'start_service_operation', 'endpoint_name' => 'billing', 'service_name' => 'invoicing', 'operation_name' => 'createinvoice', - 'request_payload' => ['codec' => 'avro', 'blob' => 'wwHioz3/VYAiNwwCDgIOaW52b2ljZQRUAAA='], + 'request_payload' => ['codec' => 'avro', 'blob' => 'wwHioz3/VYAiNwwCChRpbnZvaWNlLTQyAA=='], ]], ], $this->workerHeaders())->assertOk(); $call = WorkflowServiceCall::query()->sole(); diff --git a/tests/Fixtures/CodecRegression/worker-service-operation-request.json b/tests/Fixtures/CodecRegression/worker-service-operation-request.json index 79bd591f..3a8c7cb2 100644 --- a/tests/Fixtures/CodecRegression/worker-service-operation-request.json +++ b/tests/Fixtures/CodecRegression/worker-service-operation-request.json @@ -11,11 +11,11 @@ "bindings": ["php", "python", "rust"], "value": { "type": "array", - "items": [{"type": "map", "entries": [{"key": "invoice", "value": {"type": "long", "value": "42"}}]}] + "items": [{"type": "string", "value": "invoice-42"}] }, "framing": { "encoding": "avro-single-object", - "wire_base64": "wwHioz3/VYAiNwwCDgIOaW52b2ljZQRUAAA=" + "wire_base64": "wwHioz3/VYAiNwwCChRpbnZvaWNlLTQyAA==" }, "failure_policy": {"operation": "round_trip", "error": null} }