diff --git a/app/Http/Controllers/Api/BridgeAdapterController.php b/app/Http/Controllers/Api/BridgeAdapterController.php index 0eeeea8a6..40393de06 100644 --- a/app/Http/Controllers/Api/BridgeAdapterController.php +++ b/app/Http/Controllers/Api/BridgeAdapterController.php @@ -5,6 +5,7 @@ use App\Support\AvroPayloadEnvelopeResolver; use App\Support\BridgeAdapterOutcomeContract; use App\Support\ControlPlaneProtocol; +use App\Support\ControlPlaneRequestContract; use App\Support\NamespaceExternalPayloadStorage; use App\Support\NamespaceWorkflowScope; use App\Support\TaskQueueRoutingGate; @@ -332,7 +333,7 @@ private function updateWorkflow( 'workflow_id' => ['required', 'string', 'max:128', 'regex:/^[a-zA-Z0-9._:-]+$/'], 'update_name' => ['required', 'string', 'max:255'], 'wait_for' => ['nullable', 'string', 'in:accepted,completed'], - 'wait_timeout_seconds' => ['nullable', 'integer', 'min:1'], + 'wait_timeout_seconds' => ['nullable', 'integer', 'min:1', 'max:'.ControlPlaneRequestContract::UPDATE_WAIT_TIMEOUT_SECONDS_MAX], ]); if ($targetValidator->fails()) { diff --git a/app/Http/Controllers/Api/ServiceCatalogController.php b/app/Http/Controllers/Api/ServiceCatalogController.php index 8f2306634..2f927b297 100644 --- a/app/Http/Controllers/Api/ServiceCatalogController.php +++ b/app/Http/Controllers/Api/ServiceCatalogController.php @@ -11,6 +11,7 @@ use App\Support\SearchAttributeValueValidator; use App\Support\ServiceCallAdmission; use App\Support\ServiceCallBoundary; +use App\Support\ServiceOperationUpdateWait; use Illuminate\Http\JsonResponse; use Illuminate\Http\Request; use Illuminate\Validation\ValidationException; @@ -656,6 +657,9 @@ public function executeOperation( ], ]); + // An update-bound operation hands its wait to the update admission. + ServiceOperationUpdateWait::assertWithinCeiling($operation, $validated['wait_timeout_seconds'] ?? null); + try { $validated['payload_codec'] = PayloadCodecContract::canonicalize( $validated['payload_codec'] ?? null, diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php index d6082e44f..d1b25c6b2 100644 --- a/app/Http/Controllers/Api/WorkerController.php +++ b/app/Http/Controllers/Api/WorkerController.php @@ -33,6 +33,7 @@ use App\Support\SearchAttributeValueValidator; use App\Support\ServiceCallBoundary; use App\Support\ServiceModeTimerDispatcher; +use App\Support\ServiceOperationUpdateWait; use App\Support\StreamClosedException; use App\Support\StreamErroredException; use App\Support\StreamFullException; @@ -3195,6 +3196,14 @@ private function authorizeServiceOperationCommands(Request $request, string $nam ]); abort_unless($principal !== null && app(AuthProvider::class)->authorize($principal, 'execute_operation', $resource), 403); + // An update-bound operation hands its wait to the update admission. + ServiceOperationUpdateWait::assertWithinCeiling( + ServiceOperationUpdateWait::operation($targetNamespace, (string) $command['endpoint_name'], + (string) $command['service_name'], (string) $command['operation_name']), + $command['wait_timeout_seconds'] ?? null, + "commands.{$index}.wait_timeout_seconds", + ); + if (isset($command['search_attributes'])) { $this->searchAttributeValues->validateForNamespace( $targetNamespace, diff --git a/app/Http/Controllers/Api/WorkflowController.php b/app/Http/Controllers/Api/WorkflowController.php index 60b91f98f..12022caf8 100644 --- a/app/Http/Controllers/Api/WorkflowController.php +++ b/app/Http/Controllers/Api/WorkflowController.php @@ -6,6 +6,7 @@ use App\Http\Middleware\Authenticate; use App\Support\AvroPayloadEnvelopeResolver; use App\Support\ControlPlaneProtocol; +use App\Support\ControlPlaneRequestContract; use App\Support\ControlPlaneResponseContract; use App\Support\ControlPlaneResultMapper; use App\Support\ExternalPayloadEnvelopeService; @@ -750,7 +751,7 @@ public function update(Request $request, string $workflowId, string $updateName) 'input' => ['nullable', 'array'], 'request_id' => ['nullable', 'string', 'max:255'], 'wait_for' => ['nullable', 'string', 'in:accepted,completed'], - 'wait_timeout_seconds' => ['nullable', 'integer', 'min:1'], + 'wait_timeout_seconds' => ['nullable', 'integer', 'min:1', 'max:'.ControlPlaneRequestContract::UPDATE_WAIT_TIMEOUT_SECONDS_MAX], ]); $this->rejectLegacyUpdateFields($request); diff --git a/app/Providers/AppServiceProvider.php b/app/Providers/AppServiceProvider.php index 739cf5cbc..8a940502d 100644 --- a/app/Providers/AppServiceProvider.php +++ b/app/Providers/AppServiceProvider.php @@ -30,6 +30,7 @@ use App\Support\ServiceModeBusDispatcher; use App\Support\SharedServiceBoundaryPolicy; use App\Support\StoragePressure; +use App\Support\UpdateWaitConfiguration; use App\Support\ValidatedExternalWorkflowUpdateAdmission; use App\Support\WorkflowMemoRollingCompatibility; use App\Support\WorkflowObservability; @@ -204,6 +205,7 @@ public function boot(): void WorkflowTaskLeaseConfiguration::apply(); ActivityTaskLeaseConfiguration::apply(); + UpdateWaitConfiguration::apply(); WorkflowMemoRollingCompatibility::register(); config([ diff --git a/app/Support/ControlPlaneRequestContract.php b/app/Support/ControlPlaneRequestContract.php index a44e54acd..4f3e86a89 100644 --- a/app/Support/ControlPlaneRequestContract.php +++ b/app/Support/ControlPlaneRequestContract.php @@ -8,6 +8,13 @@ final class ControlPlaneRequestContract public const VERSION = 1; + /** + * The longest `wait_timeout_seconds` an update caller may request. A + * completion wait holds the request open, so the server bounds it; a + * longer request is refused before the update is admitted. + */ + public const UPDATE_WAIT_TIMEOUT_SECONDS_MAX = 60; + /** * @return array{ * schema: string, @@ -91,6 +98,10 @@ public static function manifest(): array 'wait_for' => [ 'canonical_values' => ['accepted', 'completed'], ], + 'wait_timeout_seconds' => [ + 'minimum' => 1, + 'maximum' => self::UPDATE_WAIT_TIMEOUT_SECONDS_MAX, + ], ], 'removed_fields' => [ 'wait_policy' => 'Use wait_for.', diff --git a/app/Support/ServerWorkflowControlPlane.php b/app/Support/ServerWorkflowControlPlane.php index caa2da4fe..6ea57e0f4 100644 --- a/app/Support/ServerWorkflowControlPlane.php +++ b/app/Support/ServerWorkflowControlPlane.php @@ -3,6 +3,7 @@ namespace App\Support; use App\Contracts\RuntimeSignalControlPlane; +use Illuminate\Validation\ValidationException; use Throwable; use Workflow\Serializers\Serializer; use Workflow\V2\CommandContext; @@ -126,6 +127,20 @@ public function query(string $instanceId, string $name, array $options = []): ar public function update(string $instanceId, string $name, array $options = []): array { + // Every update admission waits at most the update ceiling, whichever + // route reached it (Server #775). The routes refuse a longer wait + // with their own validation first; this is the boundary a service + // operation's update binding reaches after its catalog lookup. + $seconds = $options['wait_timeout_seconds'] ?? null; + if (is_numeric($seconds) && (int) $seconds > ControlPlaneRequestContract::UPDATE_WAIT_TIMEOUT_SECONDS_MAX) { + throw ValidationException::withMessages([ + 'wait_timeout_seconds' => [sprintf( + 'The wait timeout seconds field must not be greater than %d.', + ControlPlaneRequestContract::UPDATE_WAIT_TIMEOUT_SECONDS_MAX, + )], + ]); + } + return $this->mutations->run( fn (): array => $this->inner->update($instanceId, $name, $options), ); diff --git a/app/Support/ServiceOperationUpdateWait.php b/app/Support/ServiceOperationUpdateWait.php new file mode 100644 index 000000000..5873a2c1d --- /dev/null +++ b/app/Support/ServiceOperationUpdateWait.php @@ -0,0 +1,80 @@ +handler_binding_kind)), ['workflow_update', 'update_workflow'], true); + } + + /** + * Refuses a wait above the ceiling for an update-bound operation, with the + * validation message the update route gives. `$path` is the dotted + * attribute the caller sent the field as. + */ + public static function assertWithinCeiling(?WorkflowServiceOperation $operation, mixed $seconds, string $path = 'wait_timeout_seconds'): void + { + if ($seconds === null || ! self::bindsWorkflowUpdate($operation)) { + return; + } + + $data = []; + data_set($data, $path, $seconds); + + // A nested command field is named as workflow-task completion + // validation names its wildcard fields, underscores kept. + Validator::make($data, [ + $path => ['integer', 'max:'.ControlPlaneRequestContract::UPDATE_WAIT_TIMEOUT_SECONDS_MAX], + ], [], str_contains($path, '.') ? [$path => $path] : [])->validate(); + } + + /** + * The catalog operation a workflow-task `start_service_operation` + * command names, in its target namespace. + */ + public static function operation(string $namespace, string $endpointName, string $serviceName, string $operationName): ?WorkflowServiceOperation + { + $endpoint = WorkflowServiceEndpoint::query() + ->where('namespace', $namespace) + ->where('endpoint_name', strtolower(trim($endpointName))) + ->first(); + + if ($endpoint === null) { + return null; + } + + $service = WorkflowService::query() + ->where('namespace', $namespace) + ->where('workflow_service_endpoint_id', $endpoint->id) + ->where('service_name', strtolower(trim($serviceName))) + ->first(); + + if ($service === null) { + return null; + } + + return WorkflowServiceOperation::query() + ->where('namespace', $namespace) + ->where('workflow_service_id', $service->id) + ->where('operation_name', strtolower(trim($operationName))) + ->first(); + } +} diff --git a/app/Support/UpdateWaitConfiguration.php b/app/Support/UpdateWaitConfiguration.php new file mode 100644 index 000000000..5b1a65284 --- /dev/null +++ b/app/Support/UpdateWaitConfiguration.php @@ -0,0 +1,30 @@ + $effective]); + + return $effective; + } +} diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 471dda2a4..755b075e7 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -163,7 +163,12 @@ any other unknown path or method answers PHP's own 404 or 405 - Durable workflow stream items and close. - Service endpoints. (The service catalog reads from Server 2.11 are served: native never holds catalog rows, so every list is PHP's empty page and every - identity its not-found answer; the `service-catalog` fixture checks 37 cases.) + identity its not-found answer; the `service-catalog` fixture checks 37 cases. + Native neither executes service operations nor accepts the workflow-task + `start_service_operation` command, which it refuses as + `unsupported_workflow_command`. When it does, an operation bound to a + workflow update must refuse a `wait_timeout_seconds` above the update + ceiling of 60 before it runs, as PHP does (Server #775).) - Bridge adapters. - (Caller-indexed Nexus operation reads are served, with the `nexus-operations` fixture. Native writes no service calls and takeover diff --git a/docs/server-reference.md b/docs/server-reference.md index 3c2d9b89b..b2eb8faa6 100644 --- a/docs/server-reference.md +++ b/docs/server-reference.md @@ -712,6 +712,18 @@ requests the retained completion result. This PHP admission correction is tracked in [#493](https://github.com/durable-workflow/server/issues/493); older affected releases can return HTTP 500 when a key is reused for another name. +A completion wait holds the request open, so the server bounds it. An update's +`wait_timeout_seconds` must be an integer from 1 to 60, on the update route and +on bridge `update_workflow` targets. A larger value is refused with HTTP 422 +(`validation_failed`) before the update is admitted, and nothing is recorded. +When the field is absent, the wait lasts +`DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS`, which is also capped at 60. If the update is still +incomplete when the wait ends, the response reports `wait_timed_out: true`; +the update itself continues, and its status can be read later. The control-plane +request contract publishes the bound as +`operations.update.fields.wait_timeout_seconds.maximum` +([#775](https://github.com/durable-workflow/server/issues/775)). + Workflow starts fail closed with `409` / `reason: "task_queue_draining"` when the requested task queue has been explicitly drained and no active worker cohort remains to claim new work. The response includes @@ -1905,7 +1917,7 @@ inside the package's `config/workflows.php` via | `DW_V2_HISTORY_FAN_OUT_WARNING_THRESHOLD` | `160` | Largest recorded parallel group size at which history pressure becomes `approaching`. | | `DW_V2_HISTORY_EXPORT_SIGNING_KEY` | (unset) | Optional HMAC key authenticating history export archives. | | `DW_V2_HISTORY_EXPORT_SIGNING_KEY_ID` | (unset) | Optional key identifier recorded alongside signed exports. | -| `DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS` | `10` | Seconds the server waits for an update to reach a terminal stage. | +| `DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS` | `10` | Seconds the server waits for an update to reach a terminal stage when the request omits `wait_timeout_seconds`. Values above 60 are treated as 60; a request may ask for 1 to 60 seconds. | | `DW_V2_UPDATE_WAIT_POLL_INTERVAL_MS` | `50` | Milliseconds between update-stage polls. | | `DW_V2_GUARDRAILS_BOOT` | `warn` | Boot-time structural guardrail mode: `warn`, `fail`, or `silent`. | | `DW_V2_LIMIT_PENDING_ACTIVITIES` | `2000` | Package-level pending-activity ceiling per run. | diff --git a/rust/src/config_audit.rs b/rust/src/config_audit.rs index cedfa3881..a23e2ada0 100644 --- a/rust/src/config_audit.rs +++ b/rust/src/config_audit.rs @@ -210,8 +210,14 @@ where } } +/// The longest update completion wait, whether a caller asks for it or the +/// operator configures the default (Server #775, +/// `ControlPlaneRequestContract::UPDATE_WAIT_TIMEOUT_SECONDS_MAX`). +pub const UPDATE_WAIT_TIMEOUT_SECONDS_MAX: i64 = 60; + /// `UpdateWaitPolicy::completionTimeoutSeconds`: the package casts the /// setting with `(int)`, then takes at least one second; unset means 10. +/// The server's `UpdateWaitConfiguration` caps it at 60. pub fn update_wait_completion_timeout_seconds(lookup: F) -> i64 where F: Fn(&str) -> Option, @@ -221,7 +227,7 @@ where "DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS", Some("WORKFLOW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS"), ) - .map_or(10, |value| value.int().max(1)) + .map_or(10, |value| value.int().clamp(1, UPDATE_WAIT_TIMEOUT_SECONDS_MAX)) } /// `server.update_validation`: how long an update request waits for its @@ -916,6 +922,9 @@ mod tests { assert_eq!(read(&[("DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS", "0")]), 1); assert_eq!(read(&[("DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS", "never")]), 1); assert_eq!(read(&[("DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS", "null")]), 10); + assert_eq!(read(&[("DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS", "60")]), 60); + assert_eq!(read(&[("DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS", "61")]), 60); + assert_eq!(read(&[("WORKFLOW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS", "600")]), 60); } #[test] diff --git a/rust/src/runtime/command_requests.rs b/rust/src/runtime/command_requests.rs index b3dfd5d87..3eabae9e6 100644 --- a/rust/src/runtime/command_requests.rs +++ b/rust/src/runtime/command_requests.rs @@ -12,11 +12,16 @@ const SIGNAL_RULES: &[(&str, &[Rule])] = &[ ("dry_run", &[Nullable, Boolean]), ]; +/// `ControlPlaneRequestContract::UPDATE_WAIT_TIMEOUT_SECONDS_MAX`: the +/// longest completion wait an update caller may ask for (Server #775). A +/// longer request is refused before the update is admitted. +pub(super) const UPDATE_WAIT_TIMEOUT_SECONDS_MAX: i64 = crate::config_audit::UPDATE_WAIT_TIMEOUT_SECONDS_MAX; + const UPDATE_RULES: &[(&str, &[Rule])] = &[ ("input", &[Nullable, Array]), ("request_id", &[Nullable, Rule::String, Max(255)]), ("wait_for", &[Nullable, Rule::String, In(&["accepted", "completed"])]), - ("wait_timeout_seconds", &[Nullable, Integer, Min(1)]), + ("wait_timeout_seconds", &[Nullable, Integer, Min(1), Max(UPDATE_WAIT_TIMEOUT_SECONDS_MAX)]), ]; fn defaults(_: &str, _: Rule) -> Option<&'static str> { @@ -80,6 +85,11 @@ mod tests { json!({"wait_for":["The selected wait for is invalid."], "wait_timeout_seconds":["The wait timeout seconds field must be at least 1."]}) ); + assert!(update(&json!({"wait_for":"completed","wait_timeout_seconds":UPDATE_WAIT_TIMEOUT_SECONDS_MAX})).is_ok()); + assert_eq!( + errors(update(&json!({"wait_for":"completed","wait_timeout_seconds":UPDATE_WAIT_TIMEOUT_SECONDS_MAX + 1}))), + json!({"wait_timeout_seconds":["The wait timeout seconds field must not be greater than 60."]}) + ); assert_eq!( errors(update(&json!({"wait_policy":"accepted"}))), json!({"wait_policy":["The wait_policy field is no longer supported. Use wait_for."]}) diff --git a/rust/src/runtime/updates.rs b/rust/src/runtime/updates.rs index 8699d4ed1..9100a1d7a 100644 --- a/rust/src/runtime/updates.rs +++ b/rust/src/runtime/updates.rs @@ -379,7 +379,8 @@ where let rules: &[(&str, &[Rule])] = &[ ("request_id", &[Rule::Nullable, Rule::String, Rule::Max(255)]), ("wait_for", &[Rule::Nullable, Rule::String, Rule::In(&["accepted", "completed"])]), - ("wait_timeout_seconds", &[Rule::Nullable, Rule::Integer, Rule::Min(1)]), + ("wait_timeout_seconds", &[Rule::Nullable, Rule::Integer, Rule::Min(1), + Rule::Max(super::command_requests::UPDATE_WAIT_TIMEOUT_SECONDS_MAX)]), ]; if let Some(error) = failure_messages(rule_errors(&request, rules, |_, _| None)) { return Err(error); diff --git a/rust/src/runtime/visibility.rs b/rust/src/runtime/visibility.rs index 97341d4d3..22993f1c4 100644 --- a/rust/src/runtime/visibility.rs +++ b/rust/src/runtime/visibility.rs @@ -102,7 +102,8 @@ pub(super) fn request_contract() -> Value { json!({"schema":"durable-workflow.v2.control-plane-request.contract", "version":1, "operations":{"list":{"fields":{"status":{"canonical_values":["running","completed","failed"], "rejected_aliases":{"cancelled":"failed","terminated":"failed","pending":"running","waiting":"running"}}}}, - "update":{"fields":{"wait_for":{"canonical_values":["accepted","completed"]}}}}}) + "update":{"fields":{"wait_for":{"canonical_values":["accepted","completed"]}, + "wait_timeout_seconds":{"minimum":1,"maximum":super::command_requests::UPDATE_WAIT_TIMEOUT_SECONDS_MAX}}}}}) } fn response(body: Value) -> Value { diff --git a/rust/tests/execution/command_requests.rs b/rust/tests/execution/command_requests.rs index 3ff14270b..3ac168504 100644 --- a/rust/tests/execution/command_requests.rs +++ b/rust/tests/execution/command_requests.rs @@ -45,3 +45,55 @@ async fn updates_take_plain_input_and_return_at_acceptance() { assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{refused}"); assert_eq!(refused["errors"]["wait_policy"], json!(["The wait_policy field is no longer supported. Use wait_for."])); } + +// Server #775: a completion wait holds the request open, so the server bounds +// the caller's wait_timeout_seconds and refuses a longer one before admission. +#[tokio::test] +async fn update_wait_timeout_is_bounded_before_admission() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + update_worker(&app, "update-worker").await; + let started = start(&app, "bounded-wait").await; + open_update_wait(&app, "update-worker").await; + let (status, refused) = request(&app, "POST", "/api/workflows/bounded-wait/update/set", + json!({"input":[5],"wait_for":"completed","wait_timeout_seconds":61})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{refused}"); + assert_eq!(refused["errors"]["wait_timeout_seconds"], + json!(["The wait timeout seconds field must not be greater than 60."])); + // The run-targeted alias applies the same rule before admission. + let run_path = format!("/api/workflows/bounded-wait/runs/{}/update/set", started["run_id"].as_str().unwrap()); + let (status, refused) = request(&app, "POST", &run_path, + json!({"input":[5],"wait_for":"completed","wait_timeout_seconds":61})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{refused}"); + assert_eq!(refused["errors"]["wait_timeout_seconds"], + json!(["The wait timeout seconds field must not be greater than 60."])); + let accepted = |history: &Value| history.as_array().unwrap().iter().filter(|event| event["event_type"] == "UpdateAccepted").count(); + assert_eq!(accepted(&run_history(&app, &started["workflow_id"], &started["run_id"]).await), 0, "refused before admission"); + let (status, update) = request(&app, "POST", "/api/workflows/bounded-wait/update/set", + json!({"input":[5],"wait_for":"accepted","wait_timeout_seconds":60})).await; + assert_eq!(status, StatusCode::ACCEPTED, "{update}"); + assert_eq!(accepted(&run_history(&app, &started["workflow_id"], &started["run_id"]).await), 1); + let (status, cluster) = request(&app, "GET", "/api/cluster/info", Value::Null).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(cluster["control_plane"]["request_contract"]["operations"]["update"]["fields"]["wait_timeout_seconds"], + json!({"minimum":1,"maximum":60})); +} + +// Server #775: a service operation bound to a workflow update hands its wait +// to the update admission. Native runs no service operation, so neither the +// execute route nor the workflow-task command can carry a longer wait here. +#[tokio::test] +async fn service_operations_cannot_carry_an_update_wait_past_the_ceiling() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "worker", json!(["echo"]), json!([])).await; + start(&app, "service-caller").await; + let task = poll(&app, "worker", "workflow").await; + let (status, refused) = finish_task(&app, &task, json!([{"type":"start_service_operation","endpoint_name":"billing", + "service_name":"approvals","operation_name":"approve","wait_for":"completed","wait_timeout_seconds":86400}])).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{refused}"); + assert_eq!(refused["reason"], "unsupported_workflow_command", "{refused}"); + let (status, unserved) = request(&app, "POST", "/api/service-endpoints/billing/services/approvals/operations/approve/execute", + json!({"wait_timeout_seconds":61})).await; + assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{unserved}"); +} diff --git a/tests/Feature/BridgeAdapterControllerTest.php b/tests/Feature/BridgeAdapterControllerTest.php index 61213467f..443e1cc48 100644 --- a/tests/Feature/BridgeAdapterControllerTest.php +++ b/tests/Feature/BridgeAdapterControllerTest.php @@ -215,6 +215,53 @@ public function test_webhook_bridge_dedupes_update_commands_by_provider_event(): ->count()); } + public function test_webhook_bridge_bounds_update_wait_timeout_before_admission(): void + { + Queue::fake(); + + $start = $this->withHeaders($this->apiHeaders()) + ->postJson('/api/workflows', [ + 'workflow_id' => 'wf-bridge-update-wait-bound', + 'workflow_type' => 'tests.interactive-command-workflow', + ]); + + $start->assertCreated(); + $this->runReadyWorkflowTask((string) $start->json('run_id')); + + $event = static fn (string $key, int $seconds): array => [ + 'action' => 'update_workflow', + 'idempotency_key' => $key, + 'target' => [ + 'workflow_id' => 'wf-bridge-update-wait-bound', + 'update_name' => 'approve', + 'wait_timeout_seconds' => $seconds, + ], + 'input' => [true, 'pagerduty'], + ]; + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/bridge-adapters/webhook/pagerduty', $event('pagerduty-wait-61', 61)) + ->assertStatus(422) + ->assertJsonPath('outcome', 'rejected') + ->assertJsonPath('reason', 'malformed_payload') + ->assertJsonPath('errors.wait_timeout_seconds.0', 'The wait timeout seconds field must not be greater than 60.'); + + $this->assertSame(0, WorkflowCommand::query() + ->where('workflow_instance_id', 'wf-bridge-update-wait-bound') + ->where('command_type', 'update') + ->count()); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/bridge-adapters/webhook/pagerduty', $event('pagerduty-wait-60', 60)) + ->assertStatus(202) + ->assertJsonPath('accepted', true); + + $this->assertSame(1, WorkflowCommand::query() + ->where('workflow_instance_id', 'wf-bridge-update-wait-bound') + ->where('command_type', 'update') + ->count()); + } + public function test_webhook_bridge_rejects_json_tagged_update_before_returning_duplicate_outcome(): void { Queue::fake(); diff --git a/tests/Feature/ServiceOperationUpdateWaitTest.php b/tests/Feature/ServiceOperationUpdateWaitTest.php new file mode 100644 index 000000000..3d2a3f99b --- /dev/null +++ b/tests/Feature/ServiceOperationUpdateWaitTest.php @@ -0,0 +1,266 @@ +createNamespace('default'); + $this->configureWorkflowTypes([ + 'tests.interactive-command-workflow' => InteractiveCommandWorkflow::class, + ]); + $start = $this->withHeaders($this->apiHeaders())->postJson('/api/workflows', [ + 'workflow_id' => 'wf-update-target', + 'workflow_type' => 'tests.interactive-command-workflow', + ])->assertCreated(); + $this->runReadyWorkflowTask((string) $start->json('run_id')); + } + + public function test_the_execute_route_bounds_an_update_bound_operations_wait(): void + { + $this->seedOperation('update_workflow'); + $route = '/api/service-endpoints/billing/services/approvals/operations/approve/execute'; + $body = fn (int $seconds): array => [ + 'arguments' => [true, 'service'], + 'mode_override' => 'async', + 'wait_for' => 'completed', + 'wait_timeout_seconds' => $seconds, + ]; + + $this->withHeaders($this->apiHeaders())->postJson($route, $body(61)) + ->assertStatus(422) + ->assertJsonPath('errors.wait_timeout_seconds.0', 'The wait timeout seconds field must not be greater than 60.'); + $this->assertNothingAdmitted(); + + $this->withHeaders($this->apiHeaders())->postJson($route, [...$body(60), 'wait_for' => 'accepted']) + ->assertSuccessful() + ->assertJsonPath('accepted', true); + $this->assertSame(1, WorkflowServiceCall::query()->count()); + $this->assertSame(1, $this->targetUpdates()); + } + + public function test_the_workflow_task_command_bounds_an_update_bound_operations_wait(): void + { + $this->seedOperation('update_workflow'); + $this->registerWorker('caller-worker', 'callers', supportedWorkflowTypes: ['tests.service-caller']); + $this->withHeaders($this->apiHeaders())->postJson('/api/workflows', [ + 'workflow_id' => 'wf-service-caller', 'workflow_type' => 'tests.service-caller', + 'task_queue' => 'callers', 'input' => [], + ])->assertCreated(); + $task = $this->postJson('/api/worker/workflow-tasks/poll', [ + 'worker_id' => 'caller-worker', 'task_queue' => 'callers', + ], $this->workerHeaders())->assertOk()->json('task'); + $this->authenticateWorker(); + + $refused = $this->complete($task, 61)->assertStatus(422); + $this->assertSame(['The commands.0.wait_timeout_seconds field must not be greater than 60.'], + $refused->json('errors')['commands.0.wait_timeout_seconds'] ?? null); + $this->assertNothingAdmitted(); + + $this->complete($task, 60)->assertOk(); + $this->assertSame(1, WorkflowServiceCall::query()->count()); + $this->assertSame(1, $this->targetUpdates()); + } + + public function test_other_bindings_keep_the_service_operation_wait_contract(): void + { + $this->seedOperation('start_workflow'); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/service-endpoints/billing/services/approvals/operations/approve/execute', [ + 'arguments' => Avro::envelope(['approved' => true, 'source' => 'service']), + 'mode_override' => 'async', + 'wait_for' => 'accepted', + 'wait_timeout_seconds' => 600, + ]) + ->assertSuccessful(); + $this->assertSame(1, WorkflowServiceCall::query()->count()); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/service-endpoints/billing/services/approvals/operations/approve/execute', [ + 'wait_timeout_seconds' => 86401, + ]) + ->assertStatus(422) + ->assertJsonPath('errors.wait_timeout_seconds.0', 'The wait timeout seconds field must not be greater than 86400.'); + } + + public function test_a_binding_changed_after_the_check_still_cannot_carry_a_longer_wait(): void + { + // The catalog says start_workflow when the route checks, and a + // concurrent change makes it a workflow update before the service + // control plane resolves it. The update admission refuses the wait. + $this->seedOperation('start_workflow'); + $this->changeBindingBeforeExecution(); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/service-endpoints/billing/services/approvals/operations/approve/execute', [ + 'arguments' => [true, 'service'], + 'mode_override' => 'async', + 'wait_for' => 'completed', + 'wait_timeout_seconds' => 61, + ]); + $this->assertSame(0, $this->targetUpdates()); + + // At the ceiling the same race admits the update. + WorkflowServiceOperation::query()->get()->each(fn (WorkflowServiceOperation $operation) => $operation->forceFill([ + 'handler_binding_kind' => 'start_workflow', 'handler_target_reference' => 'tests.interactive-command-workflow', + ])->save()); + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/service-endpoints/billing/services/approvals/operations/approve/execute', [ + 'arguments' => [true, 'service'], + 'mode_override' => 'async', + 'wait_for' => 'accepted', + 'wait_timeout_seconds' => 60, + ]) + ->assertSuccessful(); + $this->assertSame(1, $this->targetUpdates()); + WorkflowCommand::query()->where('workflow_instance_id', 'wf-update-target')->where('command_type', 'update')->delete(); + + $this->registerWorker('caller-worker', 'callers', supportedWorkflowTypes: ['tests.service-caller']); + $this->withHeaders($this->apiHeaders())->postJson('/api/workflows', [ + 'workflow_id' => 'wf-service-caller', 'workflow_type' => 'tests.service-caller', + 'task_queue' => 'callers', 'input' => [], + ])->assertCreated(); + $task = $this->postJson('/api/worker/workflow-tasks/poll', [ + 'worker_id' => 'caller-worker', 'task_queue' => 'callers', + ], $this->workerHeaders())->assertOk()->json('task'); + WorkflowServiceOperation::query()->get()->each(fn (WorkflowServiceOperation $operation) => $operation->forceFill([ + 'handler_binding_kind' => 'start_workflow', 'handler_target_reference' => 'tests.interactive-command-workflow', + ])->save()); + $this->authenticateWorker(); + $updateEvents = fn (): int => WorkflowHistoryEvent::query()->whereIn('event_type', ['UpdateAccepted', 'UpdateRejected'])->count(); + $before = $updateEvents(); + $this->complete($task, 61); + + $this->assertSame(0, $this->targetUpdates()); + $this->assertSame($before, $updateEvents()); + + } + + /** + * Rebinds the service control plane so each execution first turns the + * catalog operation into a workflow update binding, as a concurrent + * catalog change would after the server's own check. + */ + private function changeBindingBeforeExecution(): void + { + $inner = $this->app->make(ServiceControlPlane::class); + $this->app->instance(ServiceControlPlane::class, new class($inner) implements ServiceControlPlane + { + public function __construct(private readonly ServiceControlPlane $inner) {} + + public function execute(string $endpointName, string $serviceName, string $operationName, array $options = []): array + { + WorkflowServiceOperation::query()->get()->each(fn (WorkflowServiceOperation $operation) => $operation->forceFill([ + 'handler_binding_kind' => 'update_workflow', + 'handler_target_reference' => 'approve', + 'handler_binding' => ['workflow_instance_id' => 'wf-update-target', 'update_name' => 'approve'], + ])->save()); + + return $this->inner->execute($endpointName, $serviceName, $operationName, $options); + } + + public function describeCall(string $serviceCallId, array $options = []): array + { + return $this->inner->describeCall($serviceCallId, $options); + } + + public function cancelCall(string $serviceCallId, array $options = []): array + { + return $this->inner->cancelCall($serviceCallId, $options); + } + }); + } + + private function assertNothingAdmitted(): void + { + $this->assertSame(0, WorkflowServiceCall::query()->count()); + $this->assertSame(0, $this->targetUpdates()); + $this->assertSame(0, WorkflowHistoryEvent::query() + ->whereIn('event_type', ['UpdateAccepted', 'UpdateRejected', 'ServiceCallStarted']) + ->count()); + } + + private function targetUpdates(): int + { + return WorkflowCommand::query() + ->where('workflow_instance_id', 'wf-update-target') + ->where('command_type', 'update') + ->count(); + } + + private function seedOperation(string $binding): void + { + $endpoint = WorkflowServiceEndpoint::query()->create(['namespace' => 'default', 'endpoint_name' => 'billing']); + $service = WorkflowService::query()->create([ + 'workflow_service_endpoint_id' => $endpoint->id, 'namespace' => 'default', 'service_name' => 'approvals', + ]); + WorkflowServiceOperation::query()->create([ + 'workflow_service_endpoint_id' => $endpoint->id, 'workflow_service_id' => $service->id, + 'namespace' => 'default', 'operation_name' => 'approve', 'operation_mode' => 'async', + 'handler_binding_kind' => $binding, + 'handler_target_reference' => $binding === 'update_workflow' ? 'approve' : 'tests.interactive-command-workflow', + 'handler_binding' => $binding === 'update_workflow' + ? ['workflow_instance_id' => 'wf-update-target', 'update_name' => 'approve'] + : ['workflow_type' => 'tests.interactive-command-workflow', 'queue' => 'targets'], + ]); + } + + private function complete(array $task, int $seconds) + { + return $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' => 'approvals', + 'operation_name' => 'approve', + 'request_payload' => Avro::envelope(['approved' => true, 'source' => 'service']), + 'mode_override' => 'async', + 'wait_for' => 'accepted', + 'wait_timeout_seconds' => $seconds, + ]], + ], $this->workerHeaders()); + } + + private function authenticateWorker(): void + { + config([ + 'server.auth.driver' => 'token', 'server.auth.backward_compatible' => false, + 'server.auth.principal_tokens' => json_encode([[ + 'token' => 'update-wait-token', 'subject' => 'update-wait-principal', + 'roles' => ['worker'], 'tenant' => 'default', + ]]), + ]); + $this->withHeader('Authorization', 'Bearer update-wait-token'); + } +} diff --git a/tests/Feature/UpdateWaitConfigurationTest.php b/tests/Feature/UpdateWaitConfigurationTest.php new file mode 100644 index 000000000..40f461b4f --- /dev/null +++ b/tests/Feature/UpdateWaitConfigurationTest.php @@ -0,0 +1,55 @@ + 600]); + + $this->assertSame(60, UpdateWaitConfiguration::apply()); + $this->assertSame(60, UpdateWaitPolicy::completionTimeoutSeconds()); + } + + public function test_a_configured_default_within_the_ceiling_is_kept(): void + { + foreach ([1, 25, 60] as $seconds) { + config([UpdateWaitConfiguration::CONFIG_KEY => $seconds]); + + $this->assertSame($seconds, UpdateWaitConfiguration::apply()); + $this->assertSame($seconds, UpdateWaitPolicy::completionTimeoutSeconds()); + } + } + + public function test_the_server_applies_the_ceiling_to_the_environment_setting_at_boot(): void + { + $previous = getenv('DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS'); + putenv('DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS=600'); + $_ENV['DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS'] = '600'; + $_SERVER['DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS'] = '600'; + + try { + $this->refreshApplication(); + + $this->assertSame(60, UpdateWaitPolicy::completionTimeoutSeconds()); + } finally { + if ($previous === false) { + putenv('DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS'); + unset($_ENV['DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS'], $_SERVER['DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS']); + } else { + putenv('DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS='.$previous); + $_ENV['DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS'] = $previous; + $_SERVER['DW_V2_UPDATE_WAIT_COMPLETION_TIMEOUT_SECONDS'] = $previous; + } + + $this->refreshApplication(); + } + } +} diff --git a/tests/Feature/WorkflowControlPlaneTest.php b/tests/Feature/WorkflowControlPlaneTest.php index 945fcc119..9e771154b 100644 --- a/tests/Feature/WorkflowControlPlaneTest.php +++ b/tests/Feature/WorkflowControlPlaneTest.php @@ -1310,6 +1310,75 @@ public function test_terminate_is_scoped_by_namespace(): void ->assertJsonPath('reason', 'instance_not_found'); } + public function test_update_wait_timeout_seconds_is_bounded_before_admission(): void + { + Queue::fake(); + + $this->configureWorkflowTypes(); + $this->createNamespace('default', 'Default namespace'); + + $start = $this->withHeaders($this->apiHeaders()) + ->postJson('/api/workflows', [ + 'workflow_id' => 'wf-update-wait-bound', + 'workflow_type' => 'tests.interactive-command-workflow', + ]); + + $start->assertCreated(); + $this->runReadyWorkflowTask((string) $start->json('run_id')); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/workflows/wf-update-wait-bound/update/approve', [ + 'input' => [true, 'api'], + 'wait_for' => 'completed', + 'wait_timeout_seconds' => 61, + ]) + ->assertStatus(422) + ->assertJsonPath('control_plane.operation', 'update') + ->assertJsonPath( + 'validation_errors.wait_timeout_seconds.0', + 'The wait timeout seconds field must not be greater than 60.', + ); + + // The run-targeted alias applies the same rule before admission. + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/workflows/wf-update-wait-bound/runs/'.$start->json('run_id').'/update/approve', [ + 'input' => [true, 'api'], + 'wait_for' => 'completed', + 'wait_timeout_seconds' => 61, + ]) + ->assertStatus(422) + ->assertJsonPath( + 'validation_errors.wait_timeout_seconds.0', + 'The wait timeout seconds field must not be greater than 60.', + ); + + $this->assertSame(0, WorkflowCommand::query() + ->where('workflow_instance_id', 'wf-update-wait-bound') + ->where('command_type', 'update') + ->count()); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/workflows/wf-update-wait-bound/update/approve', [ + 'input' => [true, 'api'], + 'wait_for' => 'accepted', + 'wait_timeout_seconds' => 60, + ]) + ->assertStatus(202) + ->assertJsonPath('wait_for', 'accepted'); + + $this->assertSame(1, WorkflowCommand::query() + ->where('workflow_instance_id', 'wf-update-wait-bound') + ->where('command_type', 'update') + ->count()); + + $this->getJson('/api/cluster/info', $this->apiHeaders()) + ->assertOk() + ->assertJsonPath('control_plane.request_contract.operations.update.fields.wait_timeout_seconds', [ + 'minimum' => 1, + 'maximum' => 60, + ]); + } + public function test_control_plane_requests_require_an_explicit_v2_header_and_reject_legacy_wait_policy_fields(): void { Queue::fake();