optional-worker-pending-run-cancellation - Branch - PrefectHQ/prefect

Add worker-side cancellation for pending flow runs

#20022

Comparing

optional-worker-pending-run-cancellation

(

d2edc67

) with

main

(

fa652d4

)

Untouched: 2

Ignored: 6

Benchmarks

Passed

bench_import_prefect_flow

benches/bench_import.py

1.4 s1.4 s

bench_task_decorator

benches/bench_tasks.py

600.8 µs602.7 µs

Ignored

bench_flow_call[options1]

benches/bench_flows.py

Ignored

107.9 ms107.8 ms

bench_flow_decorator

benches/bench_flows.py

Ignored

+3%

5.5 ms5.4 ms

bench_flow_call[options0]

benches/bench_flows.py

Ignored

169.2 ms169.9 ms

bench_import_prefect

benches/bench_import.py

Ignored

1.1 s1.1 s

bench_task_call

benches/bench_tasks.py

Ignored

+53%

51.3 ms33.6 ms

bench_task_submit

benches/bench_tasks.py

Ignored

+35%

10.7 ms7.9 ms

Commits

Click on a commit to change the comparison range

Base

main

fa652d4

+0.12%

Add worker-side cancellation for pending flow runs Implement optional cancellation handling in workers for flow runs that are cancelled while still in PENDING state (before the runner starts). - Add PREFECT_WORKER_ENABLE_CANCELLATION setting (disabled by default) - Add PREFECT_WORKER_CANCELLATION_POLL_SECONDS setting (default 120s) - Subscribe to cancellation events via WebSocket for real-time detection - Fall back to polling if WebSocket connection fails - Poll for cancelling flow runs once on worker startup - Add kill_infrastructure() method to BaseWorker (raises NotImplementedError) - Implement kill_infrastructure() in KubernetesWorker to delete jobs - Only handle cancellation for flow runs with start_time=None (pending) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

9ba0c4a

2 days ago

by joshuastagner

-0.25%

Merge branch 'main' into optional-worker-pending-run-cancellation

ce07c10

2 days ago

by joshuastagner

-0.15%

Refactor worker cancellation to use FlowRunCancellingObserver Simplify worker-side cancellation handling by reusing the existing FlowRunCancellingObserver from the runner instead of implementing custom WebSocket subscription and polling logic. - Use FlowRunCancellingObserver for cancellation detection - Register/unregister flow run IDs with observer on submit/complete - Use lambda callback pattern consistent with runner implementation - Remove custom cancellation event handling code 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

3e5d8e2

14 hours ago

by joshuastagner

+0.02%

Make configuration required in KubernetesWorker.kill_infrastructure The base worker now always provides configuration when calling kill_infrastructure, so the Optional type and None check are no longer needed. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

0da644a

14 hours ago

by joshuastagner

+0.04%

Filter worker cancellation events by work pool ID Use EventAnyResourceFilter to filter cancellation events so workers only receive events for flow runs in their specific work pool. This uses the any_resource filter which has proper in-memory filtering support for WebSocket event streams. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

d2edc67

11 hours ago

by joshuastagner

Home Terms Privacy Docs