Shared fixtures: bind patch-deployment deregistration checks to the refused worker

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'));

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;