diff --git a/scripts/conformance/server-parity/schedule-catalog-contract.mjs b/scripts/conformance/server-parity/schedule-catalog-contract.mjs index 88e1d7abd..1a82da941 100644 --- a/scripts/conformance/server-parity/schedule-catalog-contract.mjs +++ b/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'); diff --git a/scripts/conformance/server-parity/schedule-policies-contract.mjs b/scripts/conformance/server-parity/schedule-policies-contract.mjs index 3f8e42905..3fbfa6898 100644 --- a/scripts/conformance/server-parity/schedule-policies-contract.mjs +++ b/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,7 +56,7 @@ 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'); @@ -57,7 +64,8 @@ export function checkSchedulePolicies(fixture, observation, workflowId) { 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'); diff --git a/scripts/conformance/server-parity/schedule-policies-probe.php b/scripts/conformance/server-parity/schedule-policies-probe.php index 511ecfbe6..f6d65c0bb 100644 --- a/scripts/conformance/server-parity/schedule-policies-probe.php +++ b/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]; } diff --git a/tests/Unit/ServerParitySchedulePoliciesContractTest.mjs b/tests/Unit/ServerParitySchedulePoliciesContractTest.mjs index aa36002c6..ccb8ce5ce 100644 --- a/tests/Unit/ServerParitySchedulePoliciesContractTest.mjs +++ b/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/); +});