CI: admit a killed author's unsent poll in the corrected clock contract

.github/workflows/server-parity.yml

@@ -56,6 +56,7 @@ on:
       - tests/Unit/ServerParityUpdateValidationContractTest.mjs
       - tests/Unit/ServerParityFixtureSelectionTest.mjs
       - tests/Unit/ServerParityUpdateValidationRetryTest.mjs
+      - tests/Unit/ServerParityKilledAuthorPollTest.mjs
       - tests/Unit/ServerParityImagePullTest.mjs
       - tests/Unit/ServerParityMessageStreamsContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
@@ -1800,7 +1801,7 @@ jobs:
           bash scripts/ci/run-composer.sh --rm -v "$PWD:/source" -w /source -- \
             install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress
           docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \
-            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs > parity-evidence/comparator-tests.txt
+            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs > parity-evidence/comparator-tests.txt
           docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \
             --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \
             "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt

scripts/conformance/server-parity/corrected-clock-contract.mjs

@@ -1,5 +1,6 @@
 import assert from 'node:assert/strict';
 import {gunzipSync, inflateSync, brotliDecompressSync} from 'node:zlib';
+import {checkUnsentKilledAuthorPoll} from './killed-author-poll.mjs';
 
 const typed = value => value === null ? {type: 'null', value: null}
   : typeof value === 'boolean' ? {type: 'boolean', value} : {type: 'int64', value: String(value)};
@@ -97,8 +98,12 @@ export function checkCorrectedClock(fixture, observation, workflowId) {
     const request = row.request_body?JSON.parse(row.request_body):null;
     if (row.client_cancelled) {
       assert.equal(row.method,'POST'); assert.equal(row.status,0); assert.equal(row.response_body,'');
-      assert.equal(row.response_bytes_base64,''); assert.equal(row.request_body_complete,true); assert.equal(row.upstream_request_started,true);
+      assert.equal(row.response_bytes_base64,'');
       assert(['/api/worker/workflow-tasks/poll','/api/worker/activity-tasks/poll','/api/worker/query-tasks/poll'].includes(row.path));
+      // The fixture declares cancel_unsent_original_polls: the killed original
+      // author's poll can end before the proxy forwards it (Server #754).
+      if (!row.upstream_request_started) return checkUnsentKilledAuthorPoll(row,requests,spec,phases.original.actor,workflowId);
+      assert.equal(row.request_body_complete,true); assert.equal(row.upstream_request_started,true);
       const current = [workflowId+':workflow:clock-wait',workflowId+':workflow:wider'];
       if (!current.includes(request.worker_id)) {
         assert.equal(spec.producer,'python'); assert.equal(phases.original.actor.author_exit_code,-9);

scripts/conformance/server-parity/corrected-marker-contract.mjs

@@ -1,5 +1,6 @@
 import assert from 'node:assert/strict';
 import {gunzipSync, inflateSync, brotliDecompressSync} from 'node:zlib';
+import {checkUnsentKilledAuthorPoll} from './killed-author-poll.mjs';
 
 const sha = /^[a-f0-9]{64}$/;
 const commit = /^[a-f0-9]{40}$/;
@@ -111,22 +112,8 @@ function wireRequests(fixture, observation, workflowId) {
       assert.equal(wire.response_body, '');
       assert.equal(wire.response_bytes_base64, '');
       if (!wire.upstream_request_started) {
-        assert.equal(spec.cancel_unsent_original_polls, true);
-        assert.equal(wire.upstream_request_started, false);
-        assert.equal(wire.request_body_complete, false);
-        assert.equal(wire.request_body, '', 'unsent bytes establish no worker authority');
-        const index=observation.corrected_markers.requests.indexOf(wire);
-        const prior=observation.corrected_markers.requests.slice(0,index);
-        const workerId=`${workflowId}:workflow:original`;
-        assert(prior.some(item=>item.path==='/api/worker/register'&&item.status===201
-          &&JSON.parse(item.request_body).worker_id===workerId), 'real original registration precedes cancellation');
-        assert(prior.some(item=>/^\/api\/worker\/workflow-tasks\/[^/]+\/complete$/.test(item.path)
-          &&item.status===200&&JSON.parse(item.request_body).lease_owner===workerId
-          &&JSON.parse(item.response_body).recorded===true), 'original durable admission precedes cancellation');
-        const retirement=observation.corrected_markers.requests.findIndex(item=>item.method==='DELETE'
-          &&item.path==='/api/worker/registrations/'+encodeURIComponent(workerId));
-        assert(retirement>index, 'unsent killed-author poll precedes explicit retirement');
-        return {...wire, request:null, response:null};
+        return checkUnsentKilledAuthorPoll(wire, observation.corrected_markers.requests, spec,
+          observation.corrected_markers.phases.original.actor, workflowId);
       }
       assert.equal(wire.request_body_complete, true);
       assert.equal(wire.upstream_request_started, true, 'actual forwarded original poll');

scripts/conformance/server-parity/corrected-markers/check-clock-readers.mjs

@@ -34,6 +34,21 @@ for(const item of record.cases){
     'cold public delivery omitted':value=>{value.corrected_markers.requests=value.corrected_markers.requests.filter(row=>row.path!=='/api/worker/workflow-tasks/poll');},
     'committed completion cancelled':value=>{value.corrected_markers.requests.find(row=>row.path.endsWith('/complete')).client_cancelled=true;},
   };
+  // A killed author's poll the proxy never forwarded (Server #754) is admitted
+  // only with no bytes, no response and the original's lifecycle around it.
+  if(original.corrected_markers.requests.some(row=>row.client_cancelled&&!row.upstream_request_started)){
+    const unsent=value=>value.corrected_markers.requests.find(row=>row.client_cancelled&&!row.upstream_request_started);
+    Object.assign(mutations,{
+      'unsent cancellation claims received body':value=>{unsent(value).request_body_complete=true;},
+      'unsent cancellation invents request authority':value=>{unsent(value).request_body='{}';},
+      'unsent cancellation invents forwarding':value=>{unsent(value).upstream_request_started=true;},
+      'unsent cancellation invents response':value=>{unsent(value).status=200;},
+      'unsent cancellation lacks original registration':value=>{value.corrected_markers.requests=value.corrected_markers.requests.filter(row=>
+        row.path!=='/api/worker/register'||JSON.parse(row.request_body).worker_id!==value.workflow_id+':workflow:original');},
+      'unsent cancellation after original retirement':value=>{value.corrected_markers.requests=value.corrected_markers.requests.filter(row=>
+        row.method!=='DELETE'||row.path!=='/api/worker/registrations/'+encodeURIComponent(value.workflow_id+':workflow:original'));},
+    });
+  }
   for(const[name,mutate]of Object.entries(mutations)){
     const corrupted=structuredClone(original);mutate(corrupted);assert.throws(()=>read(corrupted),undefined,name);
   }

scripts/conformance/server-parity/killed-author-poll.mjs

@@ -0,0 +1,38 @@
+// A poll the SIGKILLed original author had in flight can end before the
+// observing proxy forwards it. The server never saw that request, so it
+// carries no request bytes and no worker authority. It is admitted only for a
+// fixture that declares it, only for a no-result poll the original author can
+// have open, and only between the original's real registration and durable
+// admission and the original's explicit retirement. The corrected-marker and
+// corrected-clock contracts share this check.
+import assert from 'node:assert/strict';
+
+export function checkUnsentKilledAuthorPoll(wire, requests, spec, originalActor, workflowId) {
+  assert.equal(spec.producer, 'python');
+  assert.equal(spec.original_shutdown, 'SIGKILL');
+  assert.equal(spec.cancel_unsent_original_polls, true);
+  assert.equal(originalActor.author_exit_code, -9);
+  assert.equal(wire.client_cancelled, true);
+  assert.equal(wire.method, 'POST');
+  assert(spec.cancelled_original_polls.some(kind => wire.path === `/api/worker/${kind}-tasks/poll`),
+    'only killed original no-result polls can be cancelled');
+  assert.equal(wire.status, 0);
+  assert.equal(wire.response_body, '');
+  assert.equal(wire.response_bytes_base64, '');
+  assert.equal(wire.upstream_request_started, false);
+  assert.equal(wire.request_body_complete, false);
+  assert.equal(wire.request_body, '', 'unsent bytes establish no worker authority');
+  const index = requests.indexOf(wire);
+  assert(index >= 0, 'the poll is one of the recorded requests');
+  const prior = requests.slice(0, index);
+  const workerId = `${workflowId}:workflow:original`;
+  assert(prior.some(item => item.path === '/api/worker/register' && item.status === 201
+    && JSON.parse(item.request_body).worker_id === workerId), 'real original registration precedes cancellation');
+  assert(prior.some(item => /^\/api\/worker\/workflow-tasks\/[^/]+\/complete$/.test(item.path)
+    && item.status === 200 && JSON.parse(item.request_body).lease_owner === workerId
+    && JSON.parse(item.response_body).recorded === true), 'original durable admission precedes cancellation');
+  const retirement = requests.findIndex(item => item.method === 'DELETE'
+    && item.path === '/api/worker/registrations/' + encodeURIComponent(workerId));
+  assert(retirement > index, 'unsent killed-author poll precedes explicit retirement');
+  return {...wire, request: null, response: null};
+}

tests/Unit/ServerParityKilledAuthorPollTest.mjs

@@ -0,0 +1,74 @@
+// Server #754: a poll the SIGKILLed original author had in flight can end
+// before the observing proxy forwards it. The corrected-marker and
+// corrected-clock contracts admit it only through this shared check.
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {checkUnsentKilledAuthorPoll} from '../../scripts/conformance/server-parity/killed-author-poll.mjs';
+
+const workflowId = 'wf';
+const worker = `${workflowId}:workflow:original`;
+const spec = {producer: 'python', original_shutdown: 'SIGKILL', cancel_unsent_original_polls: true,
+  cancelled_original_polls: ['workflow', 'activity', 'query']};
+const actor = {author_exit_code: -9};
+
+// The shape of the failing record in run 40206210357: registration, durable
+// admission, the cancelled polls, then the original's explicit retirement.
+function record() {
+  const requests = [
+    {method: 'POST', path: '/api/worker/register', status: 201, request_body: JSON.stringify({worker_id: worker})},
+    {method: 'POST', path: '/api/worker/workflow-tasks/task-1/complete', status: 200,
+      request_body: JSON.stringify({lease_owner: worker}), response_body: JSON.stringify({recorded: true})},
+    {method: 'POST', path: '/api/worker/activity-tasks/poll', status: 0, client_cancelled: true,
+      request_body_complete: false, upstream_request_started: false, request_body: '', response_body: '', response_bytes_base64: ''},
+    {method: 'DELETE', path: `/api/worker/registrations/${encodeURIComponent(worker)}`, status: 200, request_body: ''},
+  ];
+  return {requests, poll: requests[2]};
+}
+
+const check = ({requests, poll}, overrides = {}) => checkUnsentKilledAuthorPoll(poll, requests,
+  {...spec, ...overrides.spec}, {...actor, ...overrides.actor}, workflowId);
+
+test('an unsent killed-author poll between registration, admission and retirement carries no request', () => {
+  const result = check(record());
+  assert.equal(result.request, null);
+  assert.equal(result.response, null);
+  for (const kind of ['workflow', 'query']) {
+    const value = record();
+    value.poll.path = `/api/worker/${kind}-tasks/poll`;
+    assert.doesNotThrow(() => check(value), kind);
+  }
+});
+
+test('anything the server could have seen, or any lifecycle outside the killed author, is refused', () => {
+  const cases = {
+    'received body': value => { value.poll.request_body_complete = true; },
+    'invented request authority': value => { value.poll.request_body = '{}'; },
+    'invented forwarding': value => { value.poll.upstream_request_started = true; },
+    'invented response status': value => { value.poll.status = 200; },
+    'invented response body': value => { value.poll.response_body = '{}'; },
+    'not a cancellation': value => { value.poll.client_cancelled = false; },
+    'result-bearing route': value => { value.poll.path = '/api/worker/workflow-tasks/task-1/complete'; },
+    'missing original registration': value => { value.requests.splice(0, 1); },
+    'another worker registered': value => { value.requests[0].request_body = JSON.stringify({worker_id: 'wf:workflow:other'}); },
+    'missing durable admission': value => { value.requests.splice(1, 1); },
+    'admission not recorded': value => { value.requests[1].response_body = JSON.stringify({recorded: false}); },
+    'missing retirement': value => { value.requests.splice(3, 1); },
+    'retirement before the poll': value => { value.requests.splice(1, 0, value.requests.splice(3, 1)[0]); },
+    'poll not in the record': value => { value.poll = {...value.poll}; },
+  };
+  for (const [name, mutate] of Object.entries(cases)) {
+    const value = record();
+    mutate(value);
+    assert.throws(() => check(value), undefined, name);
+  }
+  const specs = {
+    'undeclared unsent cancellation': {spec: {cancel_unsent_original_polls: false}},
+    'graceful shutdown': {spec: {original_shutdown: 'SIGTERM'}},
+    'other producer': {spec: {producer: 'php'}},
+    'author not killed': {actor: {author_exit_code: 0}},
+    'poll kind not declared': {spec: {cancelled_original_polls: ['workflow']}},
+  };
+  for (const [name, overrides] of Object.entries(specs)) {
+    assert.throws(() => check(record(), overrides), undefined, name);
+  }
+});