diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index dbbc4c62e..171ef93cb 100644 --- a/.github/workflows/server-parity.yml +++ b/.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 diff --git a/scripts/conformance/server-parity/corrected-clock-contract.mjs b/scripts/conformance/server-parity/corrected-clock-contract.mjs index 47d2125ed..f68b7d274 100644 --- a/scripts/conformance/server-parity/corrected-clock-contract.mjs +++ b/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); diff --git a/scripts/conformance/server-parity/corrected-marker-contract.mjs b/scripts/conformance/server-parity/corrected-marker-contract.mjs index 2c560c508..a2c93f488 100644 --- a/scripts/conformance/server-parity/corrected-marker-contract.mjs +++ b/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'); diff --git a/scripts/conformance/server-parity/corrected-markers/check-clock-readers.mjs b/scripts/conformance/server-parity/corrected-markers/check-clock-readers.mjs index 4237fb4e3..60e557e86 100644 --- a/scripts/conformance/server-parity/corrected-markers/check-clock-readers.mjs +++ b/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); } diff --git a/scripts/conformance/server-parity/killed-author-poll.mjs b/scripts/conformance/server-parity/killed-author-poll.mjs new file mode 100644 index 000000000..db4e2b45e --- /dev/null +++ b/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}; +} diff --git a/tests/Unit/ServerParityKilledAuthorPollTest.mjs b/tests/Unit/ServerParityKilledAuthorPollTest.mjs new file mode 100644 index 000000000..e00bf0bbc --- /dev/null +++ b/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); + } +});