Stop the schedule fixture probe from applying a published PHP write twice

scripts/conformance/server-parity/schedule-catalog-contract.mjs

@@ -34,7 +34,7 @@ export function checkScheduleCatalog(fixture, observation, workflowId) {
       'embedded schedule surface explicitly inapplicable');
     return {scope: 'schedule_catalog'};
   }
-  assert.ok(Number.isInteger(state.attempts) && state.attempts >= 1 && state.attempts <= 3, 'recorded sequence attempts');
+  assert.ok(Number.isInteger(state.attempts) && state.attempts >= 1 && state.attempts <= 5, 'recorded sequence attempts');
   // A sequence replayed after PHP's SQLite lock defect runs in a new namespace.
   const namespace = `${workflowId}-catalog`.toLowerCase() + (state.attempts === 1 ? '' : `-r${state.attempts}`);
   assert.deepStrictEqual([state.applicable, state.namespace], [true, namespace], 'schedule catalog namespace');

scripts/conformance/server-parity/schedule-policies-contract.mjs

@@ -41,6 +41,13 @@ const POLICY_OUTCOMES = {
   terminate_other: ['triggered', 'triggered', 'triggered'],
 };
 
+// Each step whose status differs, with its response, so a failed run names
+// the cause instead of only the position.
+function statusDifferences(steps) {
+  return steps.flatMap((step, index) => step.status === STATUSES[index] ? [] :
+    [`step ${index} ${step.request}: ${step.status}, expected ${STATUSES[index]}: ${JSON.stringify(step.response).slice(0, 400)}`]);
+}
+
 export function checkSchedulePolicies(fixture, observation, workflowId) {
   const state = observation.schedule_policies;
   assert.ok(state && typeof state === 'object', 'actual schedule policy observation');
@@ -49,15 +56,16 @@ export function checkSchedulePolicies(fixture, observation, workflowId) {
       'embedded schedule surface explicitly inapplicable');
     return {scope: 'schedule_policies'};
   }
-  assert.ok(Number.isInteger(state.attempts) && state.attempts >= 1 && state.attempts <= 3, 'recorded sequence attempts');
+  assert.ok(Number.isInteger(state.attempts) && state.attempts >= 1 && state.attempts <= 5, 'recorded sequence attempts');
   // A sequence replayed after PHP's SQLite lock defect runs in a new namespace.
   const namespace = `${workflowId}-schedules`.toLowerCase() + (state.attempts === 1 ? '' : `-r${state.attempts}`);
   assert.deepStrictEqual([state.applicable, state.namespace], [true, namespace], 'schedule policy namespace');
   assert.ok(Number.isInteger(state.retries) && state.retries >= 0, 'recorded lock retries');
   const raw = state.steps;
   const names = {[namespace]: '@namespace', [state.queue]: '@queue'};
   const steps = raw.map(step => ({request: step.request, status: step.status, response: normalized(step.response, names)}));
-  assert.deepStrictEqual(steps.map(step => step.status), STATUSES, 'schedule step statuses');
+  assert.deepStrictEqual(steps.map(step => step.status), STATUSES,
+    `schedule step statuses (${steps.length} of ${STATUSES.length} steps)\n${statusDifferences(steps).join('\n')}`);
   const at = request => steps.filter(step => step.request === request);
   // Laravel names expanded attributes raw and others with spaces.
   assert.equal(steps[1].response.message, 'The spec.timezone field must be a string. (and 9 more errors)', 'validation summary');
@@ -79,8 +87,9 @@ export function checkSchedulePolicies(fixture, observation, workflowId) {
     execution_timeout_seconds: 90, run_timeout_seconds: 30}, 'action timeout aliases');
   for (const [policy, outcomes] of Object.entries(POLICY_OUTCOMES)) {
     const id = `policy-${policy.replaceAll('_', '-')}`;
-    assert.deepStrictEqual(at(`POST schedules/${id}/trigger`).slice(0, 3).map(step => step.response.outcome), outcomes,
-      `${policy} trigger outcomes`);
+    const triggers = at(`POST schedules/${id}/trigger`).slice(0, 3).map(step => step.response);
+    assert.deepStrictEqual(triggers.map(response => response.outcome), outcomes,
+      `${policy} trigger outcomes ${JSON.stringify(triggers.map(response => response.outcome))}, expected ${JSON.stringify(outcomes)}: ${JSON.stringify(triggers).slice(0, 600)}`);
   }
   const quota = at('POST schedules/quota/trigger');
   assert.deepStrictEqual(quota.map(step => step.response.outcome), ['triggered', 'triggered'], 'two quota runs');

scripts/conformance/server-parity/schedule-policies-probe.php

@@ -12,10 +12,12 @@
 // the contract normalizes generated identities and times.
 //
 // Published PHP on SQLite can lose a write race with its own maintenance
-// roles ("database is locked"). A 500 for that rolls the whole operation
-// back, so the step is sent again. A backfill instead records the lost
-// occurrence as a failure and goes on, which changes the schedule, so the
-// whole sequence is replayed once more in a new namespace. Each is recorded.
+// roles ("database is locked"). A read is simply sent again. A write may
+// already have committed when it reports the lock, so it is sent again only
+// when its schedule reads the same as before it; otherwise, and when a
+// backfill records a lost occurrence as a failure, the whole sequence is
+// replayed in a new namespace. Native never reports this lock, so it never
+// retries or replays. Each is recorded.
 function finishHttpSchedulePolicies(Client $admin, array $fixture, string $workflowId, string $url, string $key = 'schedule_policies'): array
 {
     $definition = $fixture[$key];
@@ -40,16 +42,46 @@ function finishHttpSchedulePolicies(Client $admin, array $fixture, string $workf
         $namespace = $attempt === 1 ? $base : "$base-r$attempt";
         $admin->createNamespace($namespace, 'schedule policies λ', 30);
         $pageToken = '';
-        $send = static function (string $method, string $path, mixed $body = null) use ($http, $token, $namespace, &$retries, &$pageToken): array {
+        $request = static function (string $method, string $target, mixed $body = null) use ($http, $token, $namespace): array {
+            $response = $http->request($method, $target, ['headers' => array_filter([
+                'Authorization' => $token === null ? null : 'Bearer '.$token,
+                'Accept' => 'application/json', 'X-Namespace' => $namespace,
+                'X-Durable-Workflow-Control-Plane-Version' => '2',
+            ]), ...($body === null ? [] : ['json' => $body === [] ? (object) [] : $body])]);
+
+            return [$response->getStatusCode(), (string) $response->getBody()];
+        };
+        $lockFailure = static fn (int $status, string $text): bool => $status === 500 && str_contains($text, 'database is locked');
+        // The schedule as it reads now, to tell a lock failure that rolled
+        // back (unchanged) from one reported after its commit.
+        $snapshot = static function (?string $schedule) use ($request, $lockFailure, &$retries): ?string {
+            for ($try = 1; $schedule !== null && $try <= 5; $try++) {
+                [$status, $text] = $request('GET', 'schedules/'.rawurlencode($schedule));
+                if (! $lockFailure($status, $text)) {
+                    return "$status $text";
+                }
+                $retries++;
+                usleep(200000 * $try);
+            }
+
+            return null;
+        };
+        $send = static function (string $method, string $path, mixed $body = null) use ($request, $lockFailure, $snapshot, &$retries, &$pageToken): array {
             $target = str_replace('{token}', rawurlencode($pageToken), $path);
+            $schedule = null;
+            if ($method !== 'GET') {
+                $schedule = preg_match('#^schedules/([^/?]+)#', $path, $match) === 1
+                    ? rawurldecode($match[1])
+                    : (is_array($body) && is_string($body['schedule_id'] ?? null) ? $body['schedule_id'] : null);
+            }
+            $before = $method === 'GET' ? null : $snapshot($schedule);
             for ($try = 1; ; $try++) {
-                $response = $http->request($method, $target, ['headers' => array_filter([
-                    'Authorization' => $token === null ? null : 'Bearer '.$token,
-                    'Accept' => 'application/json', 'X-Namespace' => $namespace,
-                    'X-Durable-Workflow-Control-Plane-Version' => '2',
-                ]), ...($body === null ? [] : ['json' => $body === [] ? (object) [] : $body])]);
-                $text = (string) $response->getBody();
-                if ($response->getStatusCode() === 500 && str_contains($text, 'database is locked') && $try < 5) {
+                [$status, $text] = $request($method, $target, $body);
+                // A read is sent again; a write only when the schedule shows
+                // it rolled back. Otherwise the failure is recorded, and the
+                // recorded lock replays the sequence.
+                if ($lockFailure($status, $text) && $try < 5
+                    && ($method === 'GET' || ($before !== null && $snapshot($schedule) === $before))) {
                     $retries++;
                     usleep(200000 * $try);
 
@@ -60,34 +92,42 @@ function finishHttpSchedulePolicies(Client $admin, array $fixture, string $workf
                     $pageToken = $decoded['next_page_token'];
                 }
 
-                return ['request' => "$method $path", 'status' => $response->getStatusCode(), 'response' => $decoded];
+                return ['request' => "$method $path", 'status' => $status, 'response' => $decoded];
             }
         };
         $steps = [];
         $locked = false;
+        // Records a step; true once a response reports the lock.
+        $record = static function (array $step) use (&$steps, &$locked): bool {
+            $steps[] = $step;
+            $locked = str_contains(json_encode($step['response']), 'database is locked');
+
+            return $locked;
+        };
         foreach ($definition['steps'] as $step) {
             if ($step[0] === '@policy') {
                 $policy = $step[1];
                 $id = 'policy-'.str_replace('_', '-', $policy);
-                $steps[] = $send('POST', 'schedules', ['schedule_id' => $id, 'spec' => ['cron_expressions' => ['0 0 1 1 *']],
+                $requests = [['POST', 'schedules', ['schedule_id' => $id, 'spec' => ['cron_expressions' => ['0 0 1 1 *']],
                     'action' => ['workflow_type' => $fixture['workflow_type'], 'task_queue' => $queue, 'input' => $input],
-                    'overlap_policy' => $policy]);
+                    'overlap_policy' => $policy]]];
                 for ($trigger = 0; $trigger < 3; $trigger++) {
-                    $steps[] = $send('POST', "schedules/$id/trigger", []);
+                    $requests[] = ['POST', "schedules/$id/trigger", []];
+                }
+                $requests[] = ['GET', "schedules/$id", null];
+                foreach ($requests as [$method, $path, $body]) {
+                    if ($record($send($method, $path, $body))) {
+                        break 2;
+                    }
                 }
-                $steps[] = $send('GET', "schedules/$id");
 
                 continue;
             }
-            $steps[] = $send($step[0], $step[1], array_key_exists(2, $step) ? $fill($step[2]) : null);
-            $last = end($steps);
-            if ($last['status'] !== 500 && str_contains(json_encode($last['response']), 'database is locked')) {
-                $locked = true;
-
+            if ($record($send($step[0], $step[1], array_key_exists(2, $step) ? $fill($step[2]) : null))) {
                 break;
             }
         }
-        if (! $locked || $attempt === 3) {
+        if (! $locked || $attempt === 5) {
             return ['applicable' => true, 'namespace' => $namespace, 'queue' => $queue, 'retries' => $retries,
                 'attempts' => $attempt, 'steps' => $steps];
         }

tests/Unit/ServerParitySchedulePoliciesContractTest.mjs

@@ -48,3 +48,25 @@ for (const [label, mutate] of [
     assert.throws(() => checkSchedulePolicies(fixture, observation, workflowId));
   });
 }
+test('a status difference names the step and its response', () => {
+  const observation = model();
+  observation.schedule_policies.steps[47] = {request: 'POST schedules/quota/trigger', status: 404,
+    response: {reason: 'schedule_not_found', schedule_id: 'quota'}};
+  assert.throws(() => checkSchedulePolicies(fixture, observation, workflowId),
+    /step 47 POST schedules\/quota\/trigger: 404, expected 200: \{"reason":"schedule_not_found","schedule_id":"quota"\}/);
+});
+test('an outcome difference shows the trigger responses', () => {
+  const observation = model();
+  Object.assign(at(observation, 'POST schedules/policy-skip/trigger')[0].response, {outcome: 'skipped', reason: 'overlap_policy_skip'});
+  assert.throws(() => checkSchedulePolicies(fixture, observation, workflowId),
+    /skip trigger outcomes \["skipped","skipped","skipped"\], expected \["triggered","skipped","skipped"\]: .*"reason":"overlap_policy_skip"/);
+});
+test('a fifth sequence in its own namespace passes, a sixth does not', () => {
+  const fifth = model();
+  const replayed = JSON.parse(JSON.stringify(fifth.schedule_policies).replaceAll('test-schedule-policies-schedules', 'test-schedule-policies-schedules-r5'));
+  fifth.schedule_policies = {...replayed, attempts: 5};
+  assert.deepStrictEqual(checkSchedulePolicies(fixture, fifth, workflowId), projection);
+  const sixth = model();
+  sixth.schedule_policies = {...JSON.parse(JSON.stringify(sixth.schedule_policies).replaceAll('test-schedule-policies-schedules', 'test-schedule-policies-schedules-r6')), attempts: 6};
+  assert.throws(() => checkSchedulePolicies(fixture, sixth, workflowId), /recorded sequence attempts/);
+});