From 9a5435516cfc78f4f339954cc7b4ab82dc6b07f2 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 28 Oct 2026 05:45:49 +0000 Subject: [PATCH] Accept an idle poller's capacity refusal before its worker's own shutdown #831's merge group (run 40637163461, MariaDB restarts job 122649110376) failed legacy-two-markers-completed on the published PHP record. The replacement Rust SDK worker completed its workflow task, its idle activity poller then got PHP's typed 429 long_poll_capacity_exhausted, and the worker deregistered itself. The legacy-marker contract accepted that clean-shutdown path only for query pollers; for workflow and activity pollers it required a recovered poll. A refusal leases nothing, and the published SDKs stop all their pollers together at shutdown. The fixtures' poll_capacity policy now names the kinds that may end in shutdown (allow_shutdown_after_commit: workflow, activity, query). A refusal is accepted without recovery only when that same worker withdraws after it, and after its own accepted commit; the contract's outcome counts still guard the work. This matches #816's rule for patch deployments. The workflow and activity "withdrawal without recovery" cases are now acceptances, with rejections for no withdrawal after the refusal, a withdrawal before the worker's own commit, and another worker's withdrawal. Recovery-isolation controls place the withdrawal before the refusal for every kind. Legacy-marker contract tests pass 2585/2585. Co-Authored-By: Claude Opus 5.5 --- .../server-parity/legacy-marker-contract.mjs | 10 ++++--- .../fresh-rust-marker-deduplication.json | 2 +- .../legacy-one-marker-control.json | 2 +- .../legacy-two-markers-completed.json | 2 +- .../legacy-two-markers-pending.json | 2 +- .../ServerParityLegacyMarkerContractTest.mjs | 26 +++++++++++++------ 6 files changed, 29 insertions(+), 15 deletions(-) diff --git a/scripts/conformance/server-parity/legacy-marker-contract.mjs b/scripts/conformance/server-parity/legacy-marker-contract.mjs index 8f47b4b70..806706747 100644 --- a/scripts/conformance/server-parity/legacy-marker-contract.mjs +++ b/scripts/conformance/server-parity/legacy-marker-contract.mjs @@ -19,7 +19,7 @@ export function checkLegacyMarkerArtifacts(fixture, artifacts) { assert.equal(spec.cancel_unsent_poll_on_shutdown, true, 'explicit locally cancelled, unforwarded poll policy'); assert.deepEqual(spec.poll_capacity, {status: 429, poll_status: 'long_poll_capacity_exhausted', retry_after_seconds: 1, kinds: ['workflow', 'activity', 'query'], recovery: 'same_authority_and_options', - allow_query_shutdown_after_commit: true}, 'explicit published no-task poll admission policy'); + allow_shutdown_after_commit: ['workflow', 'activity', 'query']}, 'explicit published no-task poll admission policy'); for (const selected of [spec.producer, spec.consumer]) { assert(['python', 'rust'].includes(selected.language)); const installed = artifacts.published_sdk_artifacts?.[selected.language]; @@ -268,8 +268,12 @@ export function checkLegacyMarkerDeployment(fixture, observation, workflowId) { && item.response.worker_id === worker); const committed = phase.requests.slice(0, withdrawal).some(item => item.status === 200 && [...completions(phase), ...activities(phase)].includes(item) && item.request.lease_owner === worker); - assert(recovered || kind === 'query' && spec.poll_capacity.allow_query_shutdown_after_commit && withdrawal > index && committed, - 'poll admission recovers with identical authority/options, or an idle query worker withdraws after its own accepted commit'); + // A refusal leases nothing, so an idle poller of a worker that then + // withdraws, after its own accepted commit, loses no work: the + // published SDKs stop their workflow, activity and query pollers + // together at shutdown. The outcome counts above guard the work. + assert(recovered || spec.poll_capacity.allow_shutdown_after_commit.includes(kind) && withdrawal > index && committed, + 'poll admission recovers with identical authority/options, or an idle worker withdraws after its own accepted commit'); } else if (request.status === 503) { assert.equal(request.method, 'POST'); assert(['/api/worker/workflow-tasks/poll', '/api/worker/activity-tasks/poll'].includes(request.path)); diff --git a/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/fresh-rust-marker-deduplication.json b/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/fresh-rust-marker-deduplication.json index 1ba23bef3..b75e0cecd 100644 --- a/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/fresh-rust-marker-deduplication.json +++ b/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/fresh-rust-marker-deduplication.json @@ -73,7 +73,7 @@ }, "cancel_poll_on_shutdown": ["workflow", "activity", "query"], "cancel_unsent_poll_on_shutdown": true, - "poll_capacity": {"status": 429, "poll_status": "long_poll_capacity_exhausted", "retry_after_seconds": 1, "kinds": ["workflow", "activity", "query"], "recovery": "same_authority_and_options", "allow_query_shutdown_after_commit": true} + "poll_capacity": {"status": 429, "poll_status": "long_poll_capacity_exhausted", "retry_after_seconds": 1, "kinds": ["workflow", "activity", "query"], "recovery": "same_authority_and_options", "allow_shutdown_after_commit": ["workflow", "activity", "query"]} }, "expected_events": [ "StartAccepted", diff --git a/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-one-marker-control.json b/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-one-marker-control.json index 47fc83b57..feda32316 100644 --- a/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-one-marker-control.json +++ b/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-one-marker-control.json @@ -73,7 +73,7 @@ }, "cancel_poll_on_shutdown": ["workflow", "activity", "query"], "cancel_unsent_poll_on_shutdown": true, - "poll_capacity": {"status": 429, "poll_status": "long_poll_capacity_exhausted", "retry_after_seconds": 1, "kinds": ["workflow", "activity", "query"], "recovery": "same_authority_and_options", "allow_query_shutdown_after_commit": true} + "poll_capacity": {"status": 429, "poll_status": "long_poll_capacity_exhausted", "retry_after_seconds": 1, "kinds": ["workflow", "activity", "query"], "recovery": "same_authority_and_options", "allow_shutdown_after_commit": ["workflow", "activity", "query"]} }, "expected_events": [ "StartAccepted", diff --git a/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-two-markers-completed.json b/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-two-markers-completed.json index 51fe4a2f0..ac3084286 100644 --- a/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-two-markers-completed.json +++ b/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-two-markers-completed.json @@ -73,7 +73,7 @@ }, "cancel_poll_on_shutdown": ["workflow", "activity", "query"], "cancel_unsent_poll_on_shutdown": true, - "poll_capacity": {"status": 429, "poll_status": "long_poll_capacity_exhausted", "retry_after_seconds": 1, "kinds": ["workflow", "activity", "query"], "recovery": "same_authority_and_options", "allow_query_shutdown_after_commit": true} + "poll_capacity": {"status": 429, "poll_status": "long_poll_capacity_exhausted", "retry_after_seconds": 1, "kinds": ["workflow", "activity", "query"], "recovery": "same_authority_and_options", "allow_shutdown_after_commit": ["workflow", "activity", "query"]} }, "expected_events": [ "StartAccepted", diff --git a/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-two-markers-pending.json b/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-two-markers-pending.json index 33784dce0..3bf68bfa9 100644 --- a/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-two-markers-pending.json +++ b/tests/Fixtures/ServerParityProfiles/legacy-marker-aliases/legacy-two-markers-pending.json @@ -73,7 +73,7 @@ }, "cancel_poll_on_shutdown": ["workflow", "activity", "query"], "cancel_unsent_poll_on_shutdown": true, - "poll_capacity": {"status": 429, "poll_status": "long_poll_capacity_exhausted", "retry_after_seconds": 1, "kinds": ["workflow", "activity", "query"], "recovery": "same_authority_and_options", "allow_query_shutdown_after_commit": true} + "poll_capacity": {"status": 429, "poll_status": "long_poll_capacity_exhausted", "retry_after_seconds": 1, "kinds": ["workflow", "activity", "query"], "recovery": "same_authority_and_options", "allow_shutdown_after_commit": ["workflow", "activity", "query"]} }, "expected_events": [ "StartAccepted", diff --git a/tests/Unit/ServerParityLegacyMarkerContractTest.mjs b/tests/Unit/ServerParityLegacyMarkerContractTest.mjs index ae8e6ad64..521ec8609 100644 --- a/tests/Unit/ServerParityLegacyMarkerContractTest.mjs +++ b/tests/Unit/ServerParityLegacyMarkerContractTest.mjs @@ -130,8 +130,6 @@ for (const id of ['legacy-two-markers-pending', 'legacy-two-markers-completed', for (const [name, mutate] of [ ['cancelled HTTP response', x => x.refusal.client_cancelled = true], ['transport failure', x => x.refusal.transport_error = 'connection reset'], - ['activity capacity', x => x.refusal.path = '/api/worker/activity-tasks/poll'], - ['workflow capacity', x => x.refusal.path = '/api/worker/workflow-tasks/poll'], ['completion capacity', x => x.refusal.path = '/api/worker/query-tasks/task/complete'], ['HTTP authentication refusal', x => x.refusal.status = 401], ['leased task', x => x.refusal.response.task = {task_id: 'lost-task'}], @@ -191,18 +189,30 @@ for (const id of ['legacy-two-markers-pending', 'legacy-two-markers-completed', ['refusal has another reason', x => x.refusal.response.reason = 'unauthorized'], ]) test(`${id} ${phase} ${kind}: rejects capacity ${name}`, () => { const state = capacityRecoveryModel(); - // Disable the separate query-shutdown alternative by placing withdrawal + // Disable the separate idle-shutdown alternative by placing withdrawal // before refusal, so these mutations isolate recovery evidence. - if (kind === 'query') { - state.requests.splice(state.requests.indexOf(state.withdrawal), 1); - state.requests.splice(state.requests.indexOf(state.refusal), 0, state.withdrawal); - } + state.requests.splice(state.requests.indexOf(state.withdrawal), 1); + state.requests.splice(state.requests.indexOf(state.refusal), 0, state.withdrawal); mutate(state); assert.throws(() => checkLegacyMarkerDeployment(fixture, state.raw, 'test')); }); - if (kind !== 'query') test(`${id} ${phase} ${kind}: rejects withdrawal without an admitted replacement poll`, () => { + function idleShutdownModel() { const state = capacityRecoveryModel(); state.requests.splice(state.requests.indexOf(state.recovery), 1); + return state; + } + test(`${id} ${phase} ${kind}: accepts an idle refusal then the worker's own withdrawal after its commit`, () => { + checkLegacyMarkerDeployment(fixture, idleShutdownModel().raw, 'test'); + }); + for (const [name, mutate] of [ + ['no withdrawal after the refusal', x => { + x.requests.splice(x.requests.indexOf(x.withdrawal), 1); + x.requests.splice(x.requests.indexOf(x.refusal), 0, x.withdrawal); + }], + ['withdrawal before its own commit', x => {x.requests.splice(x.requests.indexOf(x.withdrawal), 1); x.requests.unshift(x.withdrawal);}], + ['another worker withdraws', x => x.withdrawal.response.worker_id = 'other-worker'], + ]) test(`${id} ${phase} ${kind}: rejects idle refusal with ${name}`, () => { + const state = idleShutdownModel(); mutate(state); assert.throws(() => checkLegacyMarkerDeployment(fixture, state.raw, 'test')); }); }