shutdown_timeout is not honored on SIGTERM/SIGINT — in-flight jobs are force-cancelled instead of drained

[PaulM5406]

Hey @TkTech !

When a worker receives SIGTERM or SIGINT, Worker.on_signal() immediately calls manager.cancel_all():

async def on_signal(self, signum: int):
    if signum in (signal.SIGTERM, signal.SIGINT):
        if self.shutdown_event.is_set():
            ...
        self.shutdown_event.set()
        await self.manager.cancel_all()

This cancels every task the worker owns, including the per-queue _maintain_queue tasks. The cancellation propagates into the executor's async with block, so Executor.stop() runs and force-cancels all in-flight jobs (task.cancel() on AsyncExecutor, pool.shutdown(cancel_futures=True) on the pool-based ones).

Meanwhile Worker.stop() has a proper drain: it clears _queues, then waits up to shutdown_timeout seconds for the executors to empty before tearing anything down. But the signal handler never goes through stop(), so on the most common shutdown path in production (Kubernetes, systemd, Docker — all deliver SIGTERM), shutdown_timeout is effectively dead code and every running job gets killed and left for recovery to retry.

Proposed fix

I have a working branch and would like to know if you're open to a PR. The strategy:

1. on_signal() delegates to stop() so the existing drain/timeout mechanism is the single shutdown path. A second signal still forces an immediate shutdown.
2. stop() orchestrates the shutdown explicitly.
3. A small internal Executor._drain(timeout) waits for in-flight jobs without cancelling them.
4. A few things fall out of making the drain actually hold up:

• executor teardown must not block the event loop on sync jobs it can't interrupt (thread/sub-interpreter pools shut down with wait=False; the process executor reuses the SIGUSR1 mechanism from cancel() first);
• job state updates produced during shutdown (completions from the drain, final state of force-cancelled stragglers, pool futures cancelled before they started) are flushed to the database before the worker exits, instead of leaving stale RUNNING rows;
• stop() becomes idempotent, since the signal handler and the async context manager's __aexit__ can now both call it concurrently.

No public API change. The branch also adds a tests/test_shutdown.py covering the signal path, the drain on all four executors, timeout expiry, and the flush-on-shutdown behavior; the full suite passes locally.

Branch is here if you want to look at the actual diff before I open anything: PaulM5406#1

Would you be open to a PR for this? Happy to adjust the approach if you'd prefer something else.

[TkTech]

Heya Paul, thanks for looking into this. I actually fixed this a few days ago in the #58 branch so that interrupts just use stop() - 7715dd7. I'll have to take a closer look at your diff as it touches quite a bit more 👀

[PaulM5406]

Nice ! Two things I ran into while writing tests for this though, which is why my diff ended up bigger:

1. In your version cancel_all() is inside the asyncio.timeout block, so when a job outlives shutdown_timeout, the TimeoutError skips straight past it — nothing gets torn down, wait_for_shutdown() never returns and the process just hangs until SIGKILL. And since signals now go through stop(), that's the path a stuck job hits in prod. Moving cancel_all() after the timeout block isn't enough on its own by the way: pool.shutdown(cancel_futures=True) defaults to wait=True, so it blocks the event loop on the straggler it can't interrupt. That's where the wait=False / SIGUSR1 changes in my branch come from.
2. Jobs that complete near the end of the drain can be lost: ConcurrentExecutor._on_job_completed pops from self.jobs before the update is even queued, so len(executor) hits 0 and cancel_all() kills _maintain_updates with up to ~1s of updates still buffered in outgoing. The job succeeded but stays RUNNING in the DB, and recovery re-runs it. Hence the pop reordering and the final flush in stop().

The rest (idempotent stop()) just falls out of the signal handler and __aexit__ both calling it now.

Happy to rebase my branch on #58 once it lands and trim it down to just those parts + the tests — two of them should fail on your branch as-is (test_sync_job_force_cancelled_when_drain_times_out and test_cancelled_pending_future_records_update), so they're a decent way to check the scenarios.