From 45843761c9ad2c070885d457cd0c3ba51be84627 Mon Sep 17 00:00:00 2001 From: Alec Rosenbaum Date: Fri, 25 Sep 2026 15:45:49 +0000 Subject: [PATCH 1/2] Force sync workers to exit after job timeouts --- tasktiger/worker.py | 10 ++++++---- tests/tasks.py | 10 ++++++++++ tests/test_workers.py | 21 ++++++++++++++------- 3 files changed, 30 insertions(+), 11 deletions(-) diff --git a/tasktiger/worker.py b/tasktiger/worker.py index 38d4fd1c..e3ee968d 100644 --- a/tasktiger/worker.py +++ b/tasktiger/worker.py @@ -4,7 +4,6 @@ import os import random import signal -import sys import time import uuid from collections import OrderedDict @@ -832,10 +831,13 @@ def _mark_done() -> None: runner = runner_class(self.tiger) runner.on_permanent_error(task, execution) - # Exit the process with an error code if a task timed out to - # prevent any inconsistent state in case the runner requires it. + # Force the process to exit after a task times out. A normal interpreter + # shutdown can wait indefinitely for threads started by the task. if self.executor.exit_worker_on_job_timeout and has_job_timeout: - sys.exit("exiting worker due to job timeout error") + try: + os.write(2, b"exiting worker due to job timeout error\n") + finally: + os._exit(1) def _worker_run(self) -> None: """ diff --git a/tests/tasks.py b/tests/tasks.py index c6860c7b..6d493024 100644 --- a/tests/tasks.py +++ b/tests/tasks.py @@ -1,4 +1,5 @@ import json +import threading import time from math import ceil @@ -54,6 +55,15 @@ def long_task_killed(): time.sleep(DELAY * 2) +@tiger.task(hard_timeout=DELAY) +def long_task_with_blocked_thread(): + blocker = threading.Event() + # A non-daemon thread makes normal interpreter shutdown wait indefinitely. + thread = threading.Thread(target=blocker.wait, daemon=False) + thread.start() + time.sleep(DELAY * 2) + + @tiger.task(hard_timeout=DELAY * 2) def long_task_ok(): # Signal task has started diff --git a/tests/test_workers.py b/tests/test_workers.py index a4c895c9..49983eea 100644 --- a/tests/test_workers.py +++ b/tests/test_workers.py @@ -4,7 +4,6 @@ import time from multiprocessing import Process -import pytest from freezefrog import FreezeTime from tasktiger import Task, Worker @@ -15,8 +14,8 @@ from .config import DELAY from .tasks import ( exception_task, - long_task_killed, long_task_ok, + long_task_with_blocked_thread, simple_task, sleep_task, system_exit_task, @@ -182,11 +181,19 @@ def test_handles_exception(self, tiger, ensure_queues): ensure_queues(error={"default": 1}) def test_handles_timeout(self, tiger, ensure_queues): - Task(tiger, long_task_killed).delay() - worker = Worker(tiger, executor_class=SyncExecutor) - # Worker should exit to avoid any inconsistencies. - with pytest.raises(SystemExit): - worker.run(once=True, force_once=True) + Task(tiger, long_task_with_blocked_thread).delay() + worker = Process( + target=external_worker, + kwargs={"worker_kwargs": {"executor_class": SyncExecutor}}, + ) + worker.start() + worker.join(timeout=DELAY * 5) + try: + assert worker.exitcode == 1 + finally: + if worker.is_alive(): + worker.kill() + worker.join() ensure_queues(error={"default": 1}) def test_heartbeat(self, tiger): From bdb01dd119c46914dcb2f6ffd23d9f2a2eb12f47 Mon Sep 17 00:00:00 2001 From: Alec Rosenbaum Date: Fri, 25 Sep 2026 16:19:22 +0000 Subject: [PATCH 2/2] Release 0.26.2 --- CHANGELOG.md | 4 ++++ tasktiger/__init__.py | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5c15def3..85dec8d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## Version 0.26.2 + +* Force-exit sync workers after hard timeouts so interpreter shutdown cannot leave them waiting for task-created threads ([#384](https://github.com/closeio/tasktiger/pull/384)). + ## Version 0.26.1 * `Task.tasks_from_queue` leaves out tasks without a payload if `include_not_found` is false. Versions 0.23.0 to 0.26.0 raised `UnboundLocalError` or returned the previous task again ([#382](https://github.com/closeio/tasktiger/pull/382)). diff --git a/tasktiger/__init__.py b/tasktiger/__init__.py index 2a533ba4..a2031c64 100644 --- a/tasktiger/__init__.py +++ b/tasktiger/__init__.py @@ -12,7 +12,7 @@ from .tasktiger import TaskTiger, run_worker from .worker import Worker -__version__ = "0.26.1" +__version__ = "0.26.2" __all__ = [ "TaskTiger", "Worker",