diff --git a/chronis/core/base/scheduler.py b/chronis/core/base/scheduler.py index 1aa708c..11edbff 100644 --- a/chronis/core/base/scheduler.py +++ b/chronis/core/base/scheduler.py @@ -1,5 +1,6 @@ """Base scheduler abstract class.""" +import inspect import threading from abc import ABC, abstractmethod from collections.abc import Callable @@ -176,6 +177,7 @@ def create_job(self, job: JobDefinition) -> JobInfo: self._success_handler_registry[job.job_id] = job.on_success self._ensure_function_registered(job) + self._warn_if_timeout_is_not_enforceable(job) job_data = job.to_dict() try: @@ -184,6 +186,40 @@ def create_job(self, job: JobDefinition) -> JobInfo: except ValueError as e: raise JobAlreadyExistsError(str(e)) from e + def _warn_if_timeout_is_not_enforceable(self, job: JobDefinition) -> None: + """Warn when a timeout will be recorded but cannot actually be enforced. + + A timeout on a *sync* job is a verdict, not a stop button: Python cannot kill a + thread, so when it expires Chronis marks the job failed, fires on_failure and + schedules the retry — while the function keeps running, holding whatever it holds. + Users reasonably read ``timeout=300`` as "this job stops after 300s", so say plainly + that it does not. + + Async jobs are not warned about: their timeout is a real cancellation. + """ + if not job.timeout_seconds: + return + + with self._registry_lock: + func = self._job_registry.get(job.func_name) + + # Unknown here: _ensure_function_registered already warned, and we cannot tell + # whether the process that runs it has a sync or an async function. + if func is None or inspect.iscoroutinefunction(func): + return + + self.logger.warning( + "timeout_seconds is set on a sync job, where it cannot stop the job. On expiry " + "Chronis marks the job failed and retries it, but Python cannot kill the thread, " + "so the function keeps running (see get_worker_status()['abandoned_threads']). " + "To make the timeout real, give the blocking call its own timeout (e.g. " + "requests.get(..., timeout=...)), or write the job as an async function, where " + "the timeout is a genuine cancellation.", + job_id=job.job_id, + func_name=job.func_name, + timeout_seconds=job.timeout_seconds, + ) + def _ensure_function_registered(self, job: JobDefinition) -> None: """Register the job's callable, and warn when a job function is unknown here. diff --git a/docs/ARCHITECTURE_NOTES.md b/docs/ARCHITECTURE_NOTES.md index 1e2afab..2493e9d 100644 --- a/docs/ARCHITECTURE_NOTES.md +++ b/docs/ARCHITECTURE_NOTES.md @@ -76,6 +76,14 @@ Timeouts therefore mean different things by function colour, and this is documen sync job is marked failed and abandoned. Both raise `JobTimeoutError` and take the same retry / `on_failure` path. +Because `timeout=300` reads as "this job stops after 300 seconds" and on a sync job it does +not, `create_job` warns when a timeout is set on a locally-registered sync function. The +warning is the honest version of the contract — a verdict, not a stop button — and it points +at the two things that *do* stop the work: a timeout on the blocking call itself +(`requests.get(..., timeout=...)`), or writing the job as async. Async jobs are not warned +about; nor are functions unknown to this process, where we cannot tell which colour the +executing pod will find. + ### Beyond threads: execution models considered The thread model's residual problems are really two independent axes, and the candidate diff --git a/tests/unit/core/test_scheduler_validation.py b/tests/unit/core/test_scheduler_validation.py index c75b255..638ee2f 100644 --- a/tests/unit/core/test_scheduler_validation.py +++ b/tests/unit/core/test_scheduler_validation.py @@ -378,3 +378,62 @@ def test_resume_leaves_a_stale_next_run_time_alone(self): scheduler.resume_job("late") assert scheduler.storage.get_job("late")["next_run_time"] == stale + + +class TestTimeoutEnforceabilityWarning: + """`timeout=300` reads as "this job stops after 300s". On a sync job it does not. + + Chronis marks it failed and retries, but Python cannot kill the thread, so the function + runs on. Users should learn that when they set the timeout, not when threads pile up. + """ + + def _scheduler_and_warnings(self): + scheduler = PollingScheduler(storage_adapter=InMemoryStorage(), polling_interval_seconds=1) + warnings = [] + scheduler.logger.warning = lambda msg, **kwargs: warnings.append((msg, kwargs)) + return scheduler, warnings + + def test_sync_job_with_timeout_is_warned(self): + scheduler, warnings = self._scheduler_and_warnings() + scheduler.register_job_function("sync_job", lambda **kwargs: None) + + job = scheduler.every(minutes=5).config(timeout=300).run("sync_job") + + assert len(warnings) == 1 + message, context = warnings[0] + assert "cannot stop the job" in message + assert context["job_id"] == job.job_id + assert context["timeout_seconds"] == 300 + + def test_async_job_with_timeout_is_not_warned(self): + """An async timeout is a genuine cancellation — nothing to warn about.""" + scheduler, warnings = self._scheduler_and_warnings() + + async def async_job(**kwargs): + pass + + scheduler.register_job_function("async_job", async_job) + + scheduler.every(minutes=5).config(timeout=300).run("async_job") + + assert warnings == [] + + def test_sync_job_without_timeout_is_not_warned(self): + scheduler, warnings = self._scheduler_and_warnings() + scheduler.register_job_function("sync_job", lambda **kwargs: None) + + scheduler.every(minutes=5).run("sync_job") + + assert warnings == [] + + def test_unregistered_function_is_not_double_warned(self): + """We cannot know if the process that runs it has a sync or async function. + + It already got the "not registered here" warning; a guess on top would be noise. + """ + scheduler, warnings = self._scheduler_and_warnings() + + scheduler.every(minutes=5).config(timeout=300).run("lives_on_a_worker_pod") + + assert len(warnings) == 1 + assert "not registered" in warnings[0][0]