diff --git a/scripts/conformance/server-parity/patch-deployment-contract.mjs b/scripts/conformance/server-parity/patch-deployment-contract.mjs index f13c32c77..d2684b96b 100644 --- a/scripts/conformance/server-parity/patch-deployment-contract.mjs +++ b/scripts/conformance/server-parity/patch-deployment-contract.mjs @@ -10,6 +10,21 @@ export function checkPatchPollPressureResponse(response) { if (Object.hasOwn(response, 'message')) assert.equal(typeof response.message, 'string', 'published diagnostic message is text'); } +// Whether a request deregisters exactly this worker: DELETE +// /api/worker/registrations/{workerId} with the id as one percent-encoded +// route segment, as the SDKs send it. +function deregisters(request, workerId) { + const prefix = '/api/worker/registrations/'; + if (request.method !== 'DELETE' || typeof request.path !== 'string' || !request.path.startsWith(prefix)) return false; + const segment = request.path.slice(prefix.length); + if (segment === '' || segment.includes('/')) return false; + try { + return decodeURIComponent(segment) === workerId; + } catch { + return false; + } +} + // A server whose worker long-poll wait slots are full answers a workflow or // activity poll with PHP's typed, retryable 429 (WorkerPollBackpressure, the // DW_WORKER_LONG_POLL_MAX_CONCURRENT contract) before holding a wait. It @@ -26,13 +41,23 @@ function checkLongPollCapacity(item, index, phase, observation, workflowId) { assert.equal(response.task, null, 'capacity refusal leases no task'); assert.equal(response.poll_status, 'long_poll_capacity_exhausted'); assert.equal(response.reason, 'long_poll_capacity_exhausted'); - assert.equal(response.retryable, true); - assert.equal(response.task_kind, kinds[item.path]); - assert.equal(response.namespace, observation.namespace); - assert.equal(response.task_queue, observation.task_queue); - assert.ok(Number.isSafeInteger(response.retry_after_seconds) && response.retry_after_seconds > 0 - && response.retry_after_seconds <= 25, 'bounded retry hint'); - assert.equal(item.response_retry_after, String(response.retry_after_seconds), 'Retry-After header agrees with the body'); + assert.match(item.response_retry_after ?? '', /^[1-9][0-9]*$/, 'positive explicit Retry-After'); + assert.ok(Number(item.response_retry_after) <= 25, 'retry hint fits the actual published worker deadline'); + if (item.response_projection === 'read_or_poll_summary') { + // The SDK probe keeps only a poll's core fields (patch-deployment-probe.php); + // the full bytes stay in its raw sidecar. + assert.ok(Object.keys(response).every(key => ['task', 'poll_status', 'reason', 'message'].includes(key)), + 'projected capacity summary carries only core poll fields'); + if (Object.hasOwn(response, 'message')) assert.equal(typeof response.message, 'string'); + } else { + assert.equal(response.retryable, true); + assert.equal(response.task_kind, kinds[item.path]); + assert.equal(response.namespace, observation.namespace); + assert.equal(response.task_queue, observation.task_queue); + assert.ok(Number.isSafeInteger(response.retry_after_seconds) && response.retry_after_seconds > 0 + && response.retry_after_seconds <= 25, 'bounded integer retry hint'); + assert.equal(item.response_retry_after, String(response.retry_after_seconds), 'Retry-After header agrees with the body'); + } assert.equal(item.request.worker_id, `${workflowId}:${phase.phase}`); assert.equal(item.request.task_queue, observation.task_queue); assert.ok(typeof item.request.poll_request_id === 'string' && item.request.poll_request_id.length > 0); @@ -44,8 +69,8 @@ function checkLongPollCapacity(item, index, phase, observation, workflowId) { assert.deepEqual(retry.request, item.request, 'retry changes no poll authority or request fields'); return; } - assert.ok(later.some(request => request.method === 'DELETE' && request.path.startsWith('/api/worker/registrations/') && ok(request)), - 'the same poll succeeds later, or the worker stops cleanly after the refusal'); + assert.ok(later.some(request => deregisters(request, item.request.worker_id) && ok(request)), + 'the same poll succeeds later, or the refused worker itself stops cleanly after the refusal'); } function checkPhpPollPressure(item, index, phase, observation, workflowId) { @@ -259,6 +284,8 @@ export function checkPatchDeployment(fixture, observation, workflowId) { 'real published SDK registration'); assert.equal(phase.requests.filter(item => item.method === 'DELETE' && item.path.startsWith('/api/worker/registrations/')).length, 1, 'real published SDK deregistration'); + assert.ok(phase.requests.some(item => deregisters(item, `${workflowId}:replacement`)), + 'the replacement worker deregisters itself'); } else { assert.ok(phase.diagnostics.includes('worker.registered')); assert.ok(phase.diagnostics.includes('worker.stopped')); diff --git a/tests/Unit/ServerParityPublishedPatchContractTest.mjs b/tests/Unit/ServerParityPublishedPatchContractTest.mjs index ff09f9f1f..66850a4de 100644 --- a/tests/Unit/ServerParityPublishedPatchContractTest.mjs +++ b/tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -25,7 +25,7 @@ for (const language of ['rust', 'python']) for (const checkpoint of ['pending', typed_output: structuredClone(fixture.typed_value), events: structuredClone(events), decisions: [[false, false]], worker_finished: true, consumer: mode === 'embedded' ? {applicable: false, reason: 'embedded_executes_php_author_definitions'} : structuredClone(fixture.patch_deployment.consumer), requests: [{method: 'POST', path: '/api/worker/register', status: 200}, ...(checkpoint === 'pending' ? [activity] : []), - completion([{type: 'complete_workflow'}]), {method: 'DELETE', path: '/api/worker/registrations/replacement', status: 200}]}; + completion([{type: 'complete_workflow'}]), {method: 'DELETE', path: '/api/worker/registrations/test%3Areplacement', status: 200}]}; return {...structuredClone(replacement), patch_deployment: {checkpoint: fixture.patch_deployment.checkpoint, change_id: fixture.patch_deployment.change_id, original: structuredClone(original), replacement: structuredClone(replacement)}}; } @@ -53,6 +53,7 @@ for (const language of ['rust', 'python']) for (const checkpoint of ['pending', ['unrecorded registration', x => x.requests = x.requests.filter(r => r.path !== '/api/worker/register')], ['missing shutdown', x => x.worker_finished = false], ['missing deregistration', x => x.requests = x.requests.filter(r => r.method !== 'DELETE')], + ['unrelated deregistration', x => x.requests.find(r => r.method === 'DELETE').path = '/api/worker/registrations/unrelated-worker'], ['SDK refusal hidden', x => x.requests[0].status = 409], ['new activity emitted', x => x.requests.find(r => r.path.endsWith('/complete') && r.path.includes('workflow-tasks')).request.commands.push({type: 'schedule_activity'})], ['new marker emitted', x => x.requests.find(r => r.path.endsWith('/complete') && r.path.includes('workflow-tasks')).request.commands.push({type: 'record_version_marker'})], @@ -143,10 +144,39 @@ for (const language of ['rust', 'python']) for (const checkpoint of ['pending', ['other worker', requests => requests.at(-2).request.worker_id = 'other-worker'], ['missing poll identity', requests => delete requests.at(-2).request.poll_request_id], ['neither retried nor stopped', requests => requests.push(requests.splice(-2, 1)[0])], + ['only an unrelated worker stopped', requests => requests.at(-1).path = '/api/worker/registrations/unrelated-worker'], + ['the refused worker id as a prefix', requests => requests.at(-1).path = '/api/worker/registrations/test%3Areplacement-2'], + ['another route segment', requests => requests.at(-1).path = '/api/worker/registrations/test%3Areplacement/deregister'], + ['undecodable segment', requests => requests.at(-1).path = '/api/worker/registrations/test%3Areplacement%'], + ['a string retry hint', requests => requests.at(-2).response.retry_after_seconds = '1'], + ['a fractional retry hint', requests => { requests.at(-2).response.retry_after_seconds = 1.5; requests.at(-2).response_retry_after = '1.5'; }], + ['a zero retry hint', requests => { requests.at(-2).response.retry_after_seconds = 0; requests.at(-2).response_retry_after = '0'; }], ]) test(`${language} ${checkpoint}: capacity model rejects ${name}`, () => { const raw = capacityModel(); mutate(raw.patch_deployment.replacement.requests); assert.throws(() => checkPatchDeployment(fixture, raw, 'test')); }); + // The SDK probe projects poll responses to their core fields + // (patch-deployment-probe.php, `read_or_poll_summary`). + function projectedCapacityModel() { + const raw = capacityModel(); + const refusal = raw.patch_deployment.replacement.requests.at(-2); + refusal.response = {task: null, poll_status: 'long_poll_capacity_exhausted', reason: 'long_poll_capacity_exhausted', + message: 'Retry the poll after the advertised delay.'}; + refusal.response_projection = 'read_or_poll_summary'; + return raw; + } + test(`${language} ${checkpoint}: model accepts a projected long-poll capacity summary`, () => + checkPatchDeployment(fixture, projectedCapacityModel(), 'test')); + for (const [name, mutate] of [ + ['an unreviewed projected field', refusal => refusal.response.status = 'completed'], + ['a projected leased task', refusal => refusal.response.task = {task_id: 'unknown-task'}], + ['other projected pressure', refusal => refusal.response.poll_status = 'backend_lock_pressure'], + ['a missing Retry-After', refusal => refusal.response_retry_after = null], + ['an oversized Retry-After', refusal => refusal.response_retry_after = '26'], + ]) test(`${language} ${checkpoint}: projected capacity model rejects ${name}`, () => { + const raw = projectedCapacityModel(); mutate(raw.patch_deployment.replacement.requests.at(-2)); + assert.throws(() => checkPatchDeployment(fixture, raw, 'test')); + }); test(`${language} ${checkpoint}: capacity model rejects a changed retry request`, () => { const raw = capacityModel('/api/worker/workflow-tasks/poll', true); raw.patch_deployment.replacement.requests.at(-2).request.timeout_seconds = 2;