From 0a2ebd7cc39e4582ba79dd9c83b34a7f8c19873a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 27 Oct 2026 17:52:14 +0000 Subject: [PATCH 1/3] Bind patch-deployment deregistration checks to the refused worker General's review of #814 (comment 6447895621) found that the long-poll capacity tolerance accepted a later successful DELETE of any worker registration as the refused worker's clean stop, and that the replacement phase counted one deregistration without checking whose it was. Both checks now require DELETE /api/worker/registrations/{workerId} whose single route segment percent-decodes to the worker's own id (the refused poll's worker_id, or `:replacement`). The model deregisters as the SDKs do (`test%3Areplacement`). New negative controls per language and checkpoint: an unrelated worker, the worker id as a prefix, an extra route segment and an undecodable segment for the capacity refusal, and an unrelated deregistration for the replacement phase. All 20 fail against the merged contract and pass with this change; 628 model cases pass. Co-Authored-By: Claude Opus 5.5 --- .../patch-deployment-contract.mjs | 21 +++++++++++++++++-- ...ServerParityPublishedPatchContractTest.mjs | 7 ++++++- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/scripts/conformance/server-parity/patch-deployment-contract.mjs b/scripts/conformance/server-parity/patch-deployment-contract.mjs index f13c32c77..09c752c7c 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 @@ -44,8 +59,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 +274,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..8d81b7a43 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,6 +144,10 @@ 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%'], ]) test(`${language} ${checkpoint}: capacity model rejects ${name}`, () => { const raw = capacityModel(); mutate(raw.patch_deployment.replacement.requests); assert.throws(() => checkPatchDeployment(fixture, raw, 'test')); From c321b3a2526c680eea570a7f488ae9656363510b Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:05:51 +0000 Subject: [PATCH 2/3] Check long-poll capacity refusals in the SDK probe's projected shape The published-SDK patch probe keeps only a poll response's core fields (task, poll_status, workflow_id, run_id, status, reason, message) and marks it response_projection: read_or_poll_summary. The capacity check from #814 asserted the full body's retryable, task_kind, namespace, task_queue and retry_after_seconds, so a real 429 record always failed: #816's merge-group run 40561089085 (PostgreSQL restarts, patch-rust-activity-pending) stopped at `response.retryable` undefined. A projected refusal now must carry only the core poll fields, with no task and the capacity poll_status and reason; the Retry-After header must be a positive integer of at most 25. A full body still gets the full checks. New model cases: one projected acceptance and five projected rejections per language and checkpoint; 652 cases pass. Co-Authored-By: Claude Opus 5.5 --- .../patch-deployment-contract.mjs | 22 +++++++++++++------ ...ServerParityPublishedPatchContractTest.mjs | 22 +++++++++++++++++++ 2 files changed, 37 insertions(+), 7 deletions(-) diff --git a/scripts/conformance/server-parity/patch-deployment-contract.mjs b/scripts/conformance/server-parity/patch-deployment-contract.mjs index 09c752c7c..db9c11bfc 100644 --- a/scripts/conformance/server-parity/patch-deployment-contract.mjs +++ b/scripts/conformance/server-parity/patch-deployment-contract.mjs @@ -41,13 +41,21 @@ 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.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); diff --git a/tests/Unit/ServerParityPublishedPatchContractTest.mjs b/tests/Unit/ServerParityPublishedPatchContractTest.mjs index 8d81b7a43..6ba7adf72 100644 --- a/tests/Unit/ServerParityPublishedPatchContractTest.mjs +++ b/tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -152,6 +152,28 @@ for (const language of ['rust', 'python']) for (const checkpoint of ['pending', 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; From ada96969b3620af4fb79b9f62e19b5f74e9f454f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:48:43 +0000 Subject: [PATCH 3/3] Keep the integer retry hint check for full capacity bodies General's review of c321b3a2 (#816 comment 6451174736): moving the full-body checks under the projection branch dropped Number.isSafeInteger for retry_after_seconds, so a full body with retry_after_seconds "1" and Retry-After 1 passed. The full-body branch again requires a safe integer from 1 to 25 before comparing it with the header. New controls per language and checkpoint: a string, a fractional and a zero hint. The four string controls fail on c321b3a2 and pass now; 664 model cases pass. Co-Authored-By: Claude Opus 5.5 --- .../conformance/server-parity/patch-deployment-contract.mjs | 2 ++ tests/Unit/ServerParityPublishedPatchContractTest.mjs | 3 +++ 2 files changed, 5 insertions(+) diff --git a/scripts/conformance/server-parity/patch-deployment-contract.mjs b/scripts/conformance/server-parity/patch-deployment-contract.mjs index db9c11bfc..d2684b96b 100644 --- a/scripts/conformance/server-parity/patch-deployment-contract.mjs +++ b/scripts/conformance/server-parity/patch-deployment-contract.mjs @@ -54,6 +54,8 @@ function checkLongPollCapacity(item, index, phase, observation, workflowId) { 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}`); diff --git a/tests/Unit/ServerParityPublishedPatchContractTest.mjs b/tests/Unit/ServerParityPublishedPatchContractTest.mjs index 6ba7adf72..66850a4de 100644 --- a/tests/Unit/ServerParityPublishedPatchContractTest.mjs +++ b/tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -148,6 +148,9 @@ for (const language of ['rust', 'python']) for (const checkpoint of ['pending', ['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'));