Dispatch Nexus queries durably before waiting for completion

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<string, mixed>  $options
+     * @return array<string, mixed>
+     */
+    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<string, mixed>  $result
+     * @return array<string, mixed>
+     */
+    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,

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"

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

tests/Feature/NexusWorkflowQueryServiceExecutionTest.php

@@ -11,22 +11,56 @@
 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;
 use Workflow\V2\Models\WorkflowServiceOperation;
 
 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,