Merge pull request #945 from durable-workflow/fix/942-full-contract-qualification

Preserve update lock bounds and qualify service scheduling contracts

Author: rmcdaniel

.github/workflows/php.yml

@@ -780,6 +780,20 @@ jobs:
         QUEUE_CONNECTION: redis
         XDEBUG_MODE: off
 
+    - name: Run service scheduling and completion contracts
+      run: >-
+        vendor/bin/phpunit tests/Unit/V2/DefaultServiceControlPlaneWorkflowBindingTest.php
+        tests/Unit/V2/ServiceCallEffectIdentityTest.php
+        tests/Unit/V2/WebhookInputValidationTest.php
+        tests/Unit/V2/WorkflowExecutorServiceOperationTest.php
+        tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php
+        tests/Unit/V2/ServiceCallSchedulerTest.php
+        --fail-on-warning --log-junit build/test-results/pr-smoke-service-scheduling.xml
+      env:
+        DB_CONNECTION: mysql
+        QUEUE_CONNECTION: redis
+        XDEBUG_MODE: off
+
     - name: Upload pull-request MySQL smoke timing report
       if: ${{ always() && github.server_url == 'https://github.com' }}
       uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a  # v7

src/V2/Support/DefaultServiceControlPlane.php

@@ -1387,6 +1387,10 @@ private function dispatchWorkflowUpdate(
         $commandOptions['wait_for'] ??= $call->operation_mode === ServiceCallOperationMode::Sync->value
             ? 'completed'
             : 'accepted';
+        if ($call->getConnection()->transactionLevel() > 0) {
+            $commandOptions['wait_for'] = UpdateWaitPolicy::WAIT_FOR_ACCEPTED;
+            unset($commandOptions['wait_timeout_seconds']);
+        }
 
         try {
             $result = $this->workflows()

tests/Unit/V2/ServiceCallEffectIdentityTest.php

@@ -442,7 +442,7 @@ private function currentClaim(WorkflowStub $caller): array
             'handler_target_reference' => 'change',
         ]);
         $control = new DefaultServiceControlPlane(new DefaultWorkflowControlPlane());
-        DB::transaction(static fn (): array => $control->execute('billing', 'effects', 'apply', [
+        $options = ServiceCallScheduler::withAfterCommitAuthority([
             'namespace' => 'billing',
             'caller_namespace' => $run->namespace,
             'caller_workflow_instance_id' => $caller->id(),
@@ -451,7 +451,8 @@ private function currentClaim(WorkflowStub $caller): array
             'idempotency_key' => 'effect-identity-' . $caller->id(),
             'wait_for' => 'completed',
             'wait_timeout_seconds' => 60,
-        ]));
+        ]);
+        DB::transaction(static fn (): array => $control->execute('billing', 'effects', 'apply', $options));
         $claim = ServiceCallScheduler::claim(WorkflowServiceCall::query()->sole()->id);
         $this->assertNotNull($claim);
         $identity = ServiceCallEffectIdentity::forCurrentClaim($claim);

tests/Unit/V2/ServiceCallSchedulerTest.php

@@ -112,8 +112,8 @@ public function testCallerContextAndIdempotencyTransactionDoNotScheduleAnHttpExe
                 return $instanceId === 'accepted-target'
                     && $name === 'approve'
                     && WorkflowServiceCall::query()->getModel()->getConnection()->transactionLevel() === 1
-                    && $options['wait_for'] === 'completed'
-                    && $options['wait_timeout_seconds'] === 60;
+                    && $options['wait_for'] === 'accepted'
+                    && ! array_key_exists('wait_timeout_seconds', $options);
             })
             ->andReturn([
                 'accepted' => true,

tests/Unit/V2/WebhookInputValidationTest.php

@@ -81,6 +81,9 @@ public function testMalformedRequestReportsItsFieldBeforeChangingDurableRecords(
                 'name' => 'Ada',
             ], $payload);
         }
+        if (str_ends_with($path, '/complete')) {
+            $this->withHeader('X-Durable-Workflow-Protocol-Version', '1.20');
+        }
         $before = $this->records();
         $writes = [];
         DB::listen(static function (QueryExecuted $event) use (&$writes): void {

tests/Unit/V2/WorkflowExecutorServiceOperationTest.php

@@ -30,6 +30,7 @@
 use Workflow\V2\Support\RunServiceOperationAuthorizer;
 use Workflow\V2\Support\RunTimelineProjector;
 use Workflow\V2\Support\ServiceCallPrincipal;
+use Workflow\V2\Support\ServiceCallScheduler;
 use Workflow\V2\Support\ServiceOperationOptions;
 use Workflow\V2\Support\ServiceOperationResult;
 use Workflow\V2\Support\WorkerCompatibilityFleet;
@@ -138,8 +139,11 @@ public function testPublicExecutionRecordsTheBoundaryOutcomeAndReplaysWithoutCal
         }
         $this->assertInstanceOf(ServiceCallPrincipal::class, $service->executions[0][3]['principal']);
         $service->executions[0][3]['principal'] = $service->executions[0][3]['principal']->toAuditArray();
+        $this->assertTrue(ServiceCallScheduler::hasAfterCommitAuthority($service->executions[0][3]));
         $this->assertSame(
-            $this->ordered(['billing', 'invoices', 'create', $expectedOptions]),
+            $this->ordered(
+                ['billing', 'invoices', 'create', ServiceCallScheduler::withAfterCommitAuthority($expectedOptions)]
+            ),
             $this->ordered($service->executions[0])
         );
         $event = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->whereIn('event_type', [