diff --git a/app/Http/Controllers/Api/ServiceCatalogController.php b/app/Http/Controllers/Api/ServiceCatalogController.php index 2a87cf9eb..b7f186598 100644 --- a/app/Http/Controllers/Api/ServiceCatalogController.php +++ b/app/Http/Controllers/Api/ServiceCatalogController.php @@ -8,6 +8,7 @@ use App\Support\BackendLockPressure; use App\Support\ControlPlaneFailureDiagnostics; use App\Support\ControlPlaneProtocol; +use App\Support\LongPoller; use App\Support\NamespaceExternalPayloadStorage; use App\Support\PayloadCodecContract; use App\Support\SearchAttributeValueValidator; @@ -28,6 +29,7 @@ use Workflow\V2\Support\ServiceCallDeadlinePolicy; use Workflow\V2\Support\ServiceCallIdempotency; use Workflow\V2\Support\ServiceCallPrincipal; +use Workflow\V2\Support\ServiceCallScheduler; use Workflow\V2\Support\UpdateWaitPolicy; class ServiceCatalogController @@ -787,6 +789,7 @@ public function executeOperation( $options = $this->admittedExecutionOptions($request, $executionOptions, $callerNamespace); $options['service_call_id'] = $idempotentCall->id; $options['boundary_policy_outcome'] = 'accepted'; + $options = $this->hostQueryOptions($operation, $options); try { $result = $this->serviceControlPlane()->execute( @@ -812,6 +815,8 @@ public function executeOperation( ); } + $result = $this->waitForHostQuery($request, $result, $waitFor, $validated['wait_timeout_seconds'] ?? null); + return ControlPlaneProtocol::json($result, ($result['accepted'] ?? false) === true ? 200 : 409); } catch (ServiceOperationAuthorizationException) { return $this->serviceOperationForbidden(); @@ -862,6 +867,7 @@ public function executeOperation( $options = $this->admittedExecutionOptions($request, $executionOptions, $callerNamespace); $options['service_call_id'] = $admission->call->id; $options['boundary_policy_outcome'] = $admission->decision->outcome->value; + $options = $this->hostQueryOptions($operation, $options); try { if ($admission->replayed && ((string) $admission->call->status !== 'accepted' @@ -883,6 +889,7 @@ public function executeOperation( if ($admission->replayed) { $result['idempotent_replay'] = true; } + $result = $this->waitForHostQuery($request, $result, $waitFor, $validated['wait_timeout_seconds'] ?? null); return ControlPlaneProtocol::json( array_replace( @@ -906,6 +913,49 @@ public function executeOperation( } } + /** + * @param array $options + * @return array + */ + private function hostQueryOptions(WorkflowServiceOperation $operation, array $options): array + { + if (! in_array($operation->handler_binding_kind, ['query_workflow', 'workflow_query'], true)) { + return $options; + } + + // The broker may wait for a route and then for its response. Keep that + // entire worker wait inside one valid durable claim, outside SQL. + $queryTimeout = max(0, (int) config('server.query_tasks.timeout', 45)); + + return ServiceCallScheduler::withHostQueryAuthority($options, 2 * $queryTimeout + 30); + } + + /** + * @param array $result + * @return array + */ + private function waitForHostQuery(Request $request, array $result, string $waitFor, ?int $timeoutSeconds): array + { + $id = $result['service_call_id'] ?? null; + if ($waitFor !== 'completed' || ! is_string($id) + || ($result['resolved_binding_kind'] ?? null) !== 'workflow_query' + || in_array($result['status'] ?? null, ['completed', 'failed', 'cancelled'], true) + || ! ServiceCallScheduler::hasDispatch($id)) { + return $result; + } + + $observed = app(LongPoller::class)->until( + fn (): array => $this->serviceControlPlane()->describeCall($id, ['namespace' => $this->namespace($request)]), + static fn (array $call): bool => in_array($call['status'] ?? null, ['completed', 'failed', 'cancelled'], true), + $timeoutSeconds ?? UpdateWaitPolicy::completionTimeoutSeconds(), + UpdateWaitPolicy::pollIntervalMilliseconds(), + ); + + return array_replace($result, $observed, [ + 'accepted' => ! in_array($observed['status'] ?? null, ['failed', 'cancelled'], true), + ]); + } + private function serviceOperationOutcomeUnknown( Request $request, string $serviceCallId, diff --git a/composer.json b/composer.json index c44304b11..4006b5256 100644 --- a/composer.json +++ b/composer.json @@ -6,7 +6,7 @@ "require": { "php": "^8.2", "apache/avro": "^1.12", - "durable-workflow/workflow": "2.15.2", + "durable-workflow/workflow": "2.15.4", "laravel/framework": "^13.30", "laravel/tinker": "^3.0", "league/flysystem-aws-s3-v3": "^3.35.3" diff --git a/composer.lock b/composer.lock index 960e6af55..e6c1c5a2b 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": "595fb742ed769580048277d4900df5bf", + "content-hash": "431dd122c5fcd9b47bb573b2ec90c9b1", "packages": [ { "name": "apache/avro", @@ -655,16 +655,16 @@ }, { "name": "durable-workflow/workflow", - "version": "2.15.2", + "version": "2.15.4", "source": { "type": "git", "url": "https://github.com/durable-workflow/workflow.git", - "reference": "15ac2f1ccd5a2e4944bf39c31562d66a62a9d333" + "reference": "8af7a0c89be759dfd9eddeedc0f724bf4211a7cd" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/15ac2f1ccd5a2e4944bf39c31562d66a62a9d333", - "reference": "15ac2f1ccd5a2e4944bf39c31562d66a62a9d333", + "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/8af7a0c89be759dfd9eddeedc0f724bf4211a7cd", + "reference": "8af7a0c89be759dfd9eddeedc0f724bf4211a7cd", "shasum": "" }, "require": { @@ -705,7 +705,7 @@ "dev-main": "2.0.x-dev" }, "durable-workflow": { - "product-train": "2.15.2", + "product-train": "2.15.4", "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json", "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json" } @@ -732,9 +732,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.15.2" + "source": "https://github.com/durable-workflow/workflow/tree/2.15.4" }, - "time": "2026-10-27T09:27:18+00:00" + "time": "2026-10-27T17:51:02+00:00" }, { "name": "egulias/email-validator", diff --git a/docs/contracts/nexus.md b/docs/contracts/nexus.md index 2ba074b64..0956901d0 100644 --- a/docs/contracts/nexus.md +++ b/docs/contracts/nexus.md @@ -320,6 +320,22 @@ The contract is symmetric across modes: a sync caller has the same durable record as an async caller. Sync mode is a delivery preference on top of the durable record, not an alternative to it. +Workflow-query bindings persist their dispatch before contacting the worker. +`wait_for=accepted` returns that durable call immediately. `wait_for=completed` +waits outside a database transaction for up to `wait_timeout_seconds` (the +configured completion wait by default, at most 60 seconds). Zero performs an +immediate observation. When that wait ends with the query still pending, the +response remains accepted with the original `service_call_id`; inspect that call +or replay the same idempotency key to observe its result. A concurrent replay +shares the existing dispatch. Query failures use the recorded service retry +policy, and every business attempt contacts the worker separately. + +The API and PHP queue worker must share their polling cache so SDK workers can +receive queries dispatched by the queue worker. Use Redis for separate hosts. +For SQLite on one host, mount the same directory into both containers and set +`DW_POLLING_CACHE_PATH` to a path inside that shared directory. Private file +caches in separate containers cannot exchange query tasks or their results. + ## Original call deadlines An operation's `deadline_policy` is either null, a positive integer diff --git a/tests/Feature/NexusWorkflowQueryServiceExecutionTest.php b/tests/Feature/NexusWorkflowQueryServiceExecutionTest.php index a7b6218df..d70b41f7a 100644 --- a/tests/Feature/NexusWorkflowQueryServiceExecutionTest.php +++ b/tests/Feature/NexusWorkflowQueryServiceExecutionTest.php @@ -11,12 +11,16 @@ use App\Support\ServerPollingCache; use App\Support\WorkerProtocol; use App\Support\WorkflowQueryTaskBroker; -use Illuminate\Foundation\Testing\RefreshDatabase; +use Illuminate\Foundation\Testing\DatabaseMigrations; +use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; use Illuminate\Support\Str; +use PHPUnit\Framework\Attributes\DataProvider; use Tests\Feature\Concerns\ServerTestHelpers; use Tests\TestCase; use Workflow\Serializers\Serializer; +use Workflow\V2\Contracts\ServiceControlPlane; +use Workflow\V2\Jobs\RunServiceCall; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowService; use Workflow\V2\Models\WorkflowServiceEndpoint; @@ -24,9 +28,39 @@ class NexusWorkflowQueryServiceExecutionTest extends TestCase { - use RefreshDatabase; + use DatabaseMigrations; use ServerTestHelpers; + private ?string $queryDatabase = null; + + protected function beforeRefreshingDatabase(): void + { + if (config('database.default') !== 'sqlite') { + return; + } + // Query waits deliberately disconnect. Give each test a real database + // that survives reconnects without sharing a previous app's handles. + $this->queryDatabase = tempnam(sys_get_temp_dir(), 'dw-nexus-query-'); + config(['database.connections.sqlite.database' => $this->queryDatabase]); + DB::purge('sqlite'); + } + + protected function tearDown(): void + { + try { + parent::tearDown(); + } finally { + if ($this->queryDatabase !== null) { + foreach (['', '-wal', '-shm', '-journal'] as $suffix) { + $path = $this->queryDatabase.$suffix; + if (is_file($path)) { + unlink($path); + } + } + } + } + } + protected function setUp(): void { parent::setUp(); @@ -36,6 +70,40 @@ protected function setUp(): void $this->createNamespace('default'); } + #[DataProvider('pendingQueryWaits')] + public function test_accepted_or_bounded_wait_query_keeps_one_recoverable_dispatch_and_replay_identity(string $waitFor, ?int $seconds): void + { + $this->bindScriptedQueryWorker([['outcome' => 'completed', 'result' => ['maximum' => PHP_INT_MAX, 'nil' => null, 'text' => '雪']]]); + $this->registerQueryWorker('python-query-worker', 'python-queries', ['python.queryable']); + $this->startRemoteWorkflow('shared-greeter-workflow'); + $this->primeQueryTaskPoller('python-query-worker'); + $this->createWorkflowQueryOperation([]); + $body = ['wait_for' => $waitFor, 'idempotency_key' => 'durable-query-once']; + if ($seconds !== null) { + $body['wait_timeout_seconds'] = $seconds; + $this->app->forgetInstance(LongPoller::class); + } + $path = '/api/service-endpoints/greeter/services/shared/operations/greet/execute'; + $first = $this->postJson($path, $body, $this->apiHeaders()); + $first->assertOk()->assertJsonPath('accepted', true)->assertJsonPath('status', 'accepted'); + $id = $first->json('service_call_id'); + $replay = $this->postJson($path, $body, $this->apiHeaders()); + $replay->assertOk()->assertJsonPath('service_call_id', $id)->assertJsonPath('idempotent_replay', true); + Queue::assertPushed(RunServiceCall::class, 1); + Queue::pushed(RunServiceCall::class)->sole()->handle(app(ServiceControlPlane::class)); + $completed = $this->postJson($path, $body, $this->apiHeaders()); + $completed->assertOk()->assertJsonPath('service_call_id', $id)->assertJsonPath('status', 'completed'); + $this->assertSame(['maximum' => PHP_INT_MAX, 'nil' => null, 'text' => '雪'], Serializer::unserialize($completed->json('result_envelope.blob'))); + Queue::assertPushed(RunServiceCall::class, 1); + } + + public static function pendingQueryWaits(): iterable + { + yield 'accepted only' => ['accepted', null]; + yield 'zero completion wait' => ['completed', 0]; + yield 'expired bounded completion wait' => ['completed', 1]; + } + public function test_workflow_query_service_operation_retries_through_public_query_worker_and_completes(): void { $this->bindScriptedQueryWorker([ @@ -239,10 +307,38 @@ private function bindScriptedQueryWorker( ): void { $signals = app(LongPollSignalStore::class); $cache = app(ServerPollingCache::class); + $this->app->instance(LongPoller::class, new class($signals, app(LongPollWaitSlotStore::class)) extends LongPoller + { + private int $handled = 0; + + public function until( + callable $probe, + callable $ready, + ?int $timeoutSeconds = null, + ?int $intervalMilliseconds = null, + array $wakeChannels = [], + ?callable $nextProbeAt = null, + bool $reserveWorkerWaitSlot = false, + string $waitSlotPool = 'worker', + ?string $waitSlotNamespace = null, + array $interruptChannels = [], + ?callable $onInterrupt = null, + ): mixed { + $jobs = Queue::pushed(RunServiceCall::class); + while ($this->handled < $jobs->count()) { + $job = $jobs->values()->get($this->handled++); + $job->handle(app(ServiceControlPlane::class)); + $jobs = Queue::pushed(RunServiceCall::class); + } + + return $probe(); + } + }); $workerStep = function (array $task) use (&$outcomes, $namespace, $workerId, $taskQueue): void { if (($task['status'] ?? null) !== 'pending' || ! is_string($task['query_task_id'] ?? null)) { return; } + $this->assertSame(0, DB::connection()->transactionLevel(), 'Query worker completion must not inherit an HTTP admission transaction.'); $poll = $this->postJson('/api/worker/query-tasks/poll', [ 'worker_id' => $workerId,