From abfe34b108344f1ee304e3fc69423ff1aac68557 Mon Sep 17 00:00:00 2001 From: enginerd-kr Date: Sun, 12 Jul 2026 22:36:17 +0900 Subject: [PATCH 1/3] fix(core): count execution capacity in permits so a hung job cannot kill the scheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects that ARCHITECTURE_NOTES.md recorded but left unfixed, both reproduced first. A timed-out sync job permanently consumed a worker. Python cannot safely kill a thread, so a timeout marks the job failed and abandons its thread — but on a ThreadPoolExecutor that thread keeps its worker slot forever, and the pool has no way to replace it. With max_workers=2 and two hung jobs, the next job ran zero times: the node had silently stopped executing anything at all. (PR #61 fixed the queue slot; the pool slot was still gone.) And max_workers never bounded async jobs at all — they run on the event loop, not the pool. With max_workers=1, five async jobs ran concurrently. The parameter said one thing and the scheduler did another. Both come from counting capacity in threads. Capacity is now counted in permits (WorkerPool): one permit per running job, whether it runs on a worker thread or on the event loop, so max_workers finally means "jobs running at once". A job that cannot get a permit is not claimed — it stays SCHEDULED for the next poll or a node with room, rather than stranding in RUNNING until its liveness lease expires. And a timed-out job hands its permit back immediately, even though its thread lives on. The thread is still leaked; nothing can change that. But it is now counted and logged (get_worker_status()["abandoned_threads"]), so a job that hangs past its timeout shows up as a number that climbs rather than as a scheduler that mysteriously stalls. Also gives the async event loop an executor we own and name. Post-execution storage writes were pushed off the loop with run_in_executor(None, ...), which silently spins up asyncio's default pool — a fifth thread pool that nothing in Chronis configured or shut down. Timeout semantics differ by function colour and now say so on `timeout` / `timeout_seconds`: an async job is genuinely cancelled, a sync job is marked failed and abandoned. Both raise JobTimeoutError and take the same retry / on_failure path. Co-Authored-By: Claude Opus 4.8 (1M context) --- chronis/core/base/scheduler.py | 13 +- chronis/core/execution/coordinator.py | 118 +++++++----- chronis/core/execution/job_executor.py | 24 ++- chronis/core/execution/worker_pool.py | 178 ++++++++++++++++++ chronis/core/schedulers/fluent_builders.py | 8 +- chronis/core/schedulers/polling_scheduler.py | 33 +++- docs/ARCHITECTURE_NOTES.md | 44 +++-- tests/integration/test_execution_capacity.py | 149 +++++++++++++++ tests/unit/core/test_execute_queued_jobs.py | 2 +- tests/unit/core/test_worker_pool.py | 149 +++++++++++++++ .../services/test_execution_coordinator.py | 112 +++++------ 11 files changed, 696 insertions(+), 134 deletions(-) create mode 100644 chronis/core/execution/worker_pool.py create mode 100644 tests/integration/test_execution_capacity.py create mode 100644 tests/unit/core/test_worker_pool.py diff --git a/chronis/core/base/scheduler.py b/chronis/core/base/scheduler.py index 9264e32..1aa708c 100644 --- a/chronis/core/base/scheduler.py +++ b/chronis/core/base/scheduler.py @@ -13,6 +13,7 @@ ) from chronis.core.execution.callbacks import OnFailureCallback, OnSuccessCallback from chronis.core.execution.coordinator import ExecutionCoordinator +from chronis.core.execution.worker_pool import WorkerPool from chronis.core.jobs.definition import JobDefinition, JobInfo from chronis.core.schedulers.next_run_calculator import NextRunTimeCalculator from chronis.core.state import JobStatus @@ -87,23 +88,19 @@ def __init__( self._init_execution_coordinator() def _init_execution_coordinator(self) -> None: - """Build a fresh thread pool and coordinator. + """Build a fresh worker pool and coordinator. - Called again on restart: a ThreadPoolExecutor cannot be reused after shutdown, so + Called again on restart: a pool that has been shut down stops accepting work, so start() must rebuild it. The registries are shared by reference, so registered functions and handlers survive. """ - from concurrent.futures import ThreadPoolExecutor - - self._executor = ThreadPoolExecutor( - max_workers=self.max_workers, thread_name_prefix="chronis-worker-" - ) + self._workers = WorkerPool(max_workers=self.max_workers, logger=self.logger) self._executor_shutdown = False self._execution_coordinator = ExecutionCoordinator( storage=self.storage, lock=self.lock, - executor=self._executor, + workers=self._workers, function_registry=self._job_registry, failure_handler_registry=self._failure_handler_registry, success_handler_registry=self._success_handler_registry, diff --git a/chronis/core/execution/coordinator.py b/chronis/core/execution/coordinator.py index 486f67a..6b24814 100644 --- a/chronis/core/execution/coordinator.py +++ b/chronis/core/execution/coordinator.py @@ -3,7 +3,6 @@ import asyncio import threading from collections.abc import Callable, Generator -from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from dataclasses import dataclass from datetime import timedelta @@ -20,6 +19,7 @@ ) from chronis.core.execution.callbacks import OnFailureCallback, OnSuccessCallback from chronis.core.execution.job_executor import JobExecutor +from chronis.core.execution.worker_pool import Slot, WorkerPool from chronis.core.jobs.definition import JobInfo from chronis.core.schedulers.next_run_calculator import NextRunTimeCalculator from chronis.core.state import JobStatus @@ -120,7 +120,7 @@ def __init__( self, storage: JobStorageAdapter, lock: LockAdapter | None, - executor: ThreadPoolExecutor, + workers: WorkerPool, function_registry: dict[str, Callable], failure_handler_registry: dict[str, OnFailureCallback], success_handler_registry: dict[str, OnSuccessCallback], @@ -132,7 +132,7 @@ def __init__( ) -> None: self.storage = storage self.lock = lock - self.executor = executor + self.workers = workers self.function_registry = function_registry self.failure_handler_registry = failure_handler_registry self.success_handler_registry = success_handler_registry @@ -149,7 +149,11 @@ def __init__( def try_execute(self, job_data: dict[str, Any], on_complete: Callable[[str], None]) -> bool: """ - Try to execute a job with distributed lock. + Try to execute a job: reserve capacity, claim it, run it. + + Capacity is reserved *before* the job is claimed. Claiming a job we have no room to + run would strand it in RUNNING until its liveness lease expired; leaving it + SCHEDULED lets the next poll — or a node with spare capacity — take it instead. Args: job_data: Job data from storage @@ -164,27 +168,38 @@ def try_execute(self, job_data: dict[str, Any], on_complete: Callable[[str], Non job_logger = self.logger.with_context(job_id=job_id, job_name=job_name) - with self._acquire_lock_context(lock_key) as lock_acquired: - if not lock_acquired: - return False - - try: - claimed = self._try_claim_job_with_cas(job_id, job_data) + slot = self.workers.reserve() + if slot is None: + return False - if claimed is None: + started = False + try: + with self._acquire_lock_context(lock_key) as lock_acquired: + if not lock_acquired: return False - self._track_running(job_id) try: - self._trigger_execution(claimed, job_logger, on_complete) - except Exception: - self._untrack_running(job_id) - raise - return True + claimed = self._try_claim_job_with_cas(job_id, job_data) - except Exception as e: - job_logger.error("Job execution failed", error=str(e), exc_info=True) - return False + if claimed is None: + return False + + self._track_running(job_id) + try: + self._trigger_execution(claimed, job_logger, on_complete, slot) + except Exception: + self._untrack_running(job_id) + raise + started = True + return True + + except Exception as e: + job_logger.error("Job execution failed", error=str(e), exc_info=True) + return False + finally: + # Every path that did not hand the slot to a running job gives it straight back. + if not started: + slot.release() def _track_running(self, job_id: str) -> None: """Mark a job as executing on this node so heartbeat() keeps it alive.""" @@ -248,43 +263,46 @@ def _trigger_execution( claimed: ClaimedJob, job_logger: ContextLogger, on_complete: Callable[[str], None], + slot: Slot, ) -> None: """ - Trigger job execution with automatic rollback on failure. + Start the job on its slot, rolling the job back if it cannot be started at all. - Routes async functions to the shared event loop and sync functions - to the thread pool executor. + Async functions run on the shared event loop, sync functions on a worker thread. + Both hold the same kind of slot, so ``max_workers`` bounds them together. A sync job that times out cannot be killed — Python has no safe way to stop a - thread — so the timeout marks the job failed and abandons the thread. The job is - released from this node's bookkeeping at that moment rather than when the thread - eventually finishes: a thread that never finishes would otherwise hold its queue - slot forever and, because the queue rejects duplicates, block its own retry from - ever being picked up here again. + thread — so the timeout marks the job failed and abandons the thread. Everything the + job was holding is handed back at that moment rather than when the thread eventually + returns (it may never): its slot, its place in the queue, and its heartbeat. Waiting + for the thread instead would let a handful of hung jobs consume every slot and stop + the node executing anything at all. """ job_id = claimed.job_id func_name = claimed.func_name timeout_seconds = claimed.timeout_seconds - def _on_done(_future: Any) -> None: + def _finish() -> None: self._untrack_running(job_id) on_complete(job_id) + timer: threading.Timer | None = None + try: if self._job_executor.is_async(func_name): loop = self._job_executor.ensure_async_loop() future = asyncio.run_coroutine_threadsafe( self._execute_async(claimed, job_logger), loop ) - future.add_done_callback(_on_done) + + def _on_async_done(_future: Any) -> None: + slot.release() + _finish() + + future.add_done_callback(_on_async_done) else: gate = _OutcomeGate() if timeout_seconds else None - future = self.executor.submit( - self._execute_in_background, claimed, job_logger, gate - ) - future.add_done_callback(_on_done) - if timeout_seconds and gate is not None: timeout_gate = gate @@ -299,11 +317,11 @@ def _on_timeout() -> None: timeout_seconds=timeout_seconds, ) - # Hand the slot back now: the abandoned thread may never return, and - # _on_done would never fire. Both calls are idempotent, so the late - # _on_done (if it ever comes) is harmless. - self._untrack_running(job_id) - on_complete(job_id) + # Give back the capacity now. The thread is written off; its slot, + # queue place and heartbeat must not be. abandon() and _finish() are + # idempotent, so the thread returning later is harmless. + slot.abandon() + _finish() self._handle_job_failure( claimed, JobTimeoutError(f"Job '{job_id}' timed out after {timeout_seconds}s"), @@ -314,12 +332,26 @@ def _on_timeout() -> None: timer.daemon = True timer.start() - # Cancel timer when the future completes normally - future.add_done_callback(lambda f: timer.cancel()) + armed_timer = timer + + def _run() -> None: + try: + self._execute_in_background(claimed, job_logger, gate) + finally: + if armed_timer is not None: + armed_timer.cancel() + _finish() + + # Slot.run releases the slot when the thread returns — including a thread + # that was abandoned and came back later, where release() is a no-op. + slot.run(_run, name=f"chronis-job-{job_id}") except Exception as submit_error: + if timer is not None: + timer.cancel() # the job never started; don't let its timeout fire + job_logger.warning( - "Executor submit failed, rolling back to SCHEDULED", + "Could not start job, rolling back to SCHEDULED", error=str(submit_error), error_type=type(submit_error).__name__, ) diff --git a/chronis/core/execution/job_executor.py b/chronis/core/execution/job_executor.py index 8194068..b43e708 100644 --- a/chronis/core/execution/job_executor.py +++ b/chronis/core/execution/job_executor.py @@ -4,6 +4,7 @@ import inspect import threading from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor from typing import Any from chronis.core.common.exceptions import FunctionNotRegisteredError @@ -22,6 +23,7 @@ def __init__( self, function_registry: dict[str, Callable], logger: ContextLogger, + async_io_workers: int = 4, ) -> None: self.function_registry = function_registry self.logger = logger @@ -29,6 +31,8 @@ def __init__( self._async_loop: asyncio.AbstractEventLoop | None = None self._async_thread: threading.Thread | None = None self._async_lock = threading.Lock() + self._async_io_workers = async_io_workers + self._async_io_executor: ThreadPoolExecutor | None = None def is_async(self, func_name: str) -> bool: """Check if a registered function is async.""" @@ -85,11 +89,25 @@ async def execute_async(self, job_data: dict[str, Any], job_logger: ContextLogge await coro def ensure_async_loop(self) -> asyncio.AbstractEventLoop: - """Get or create the shared event loop for async jobs (thread-safe).""" + """Get or create the shared event loop for async jobs (thread-safe). + + The loop gets an executor we own and name. An async job's post-execution work + (storage writes, callbacks) is blocking, so it is pushed off the loop with + run_in_executor — which, left to itself, would silently spin up asyncio's *default* + executor: an unnamed pool, sized by a rule of thumb, that nothing in Chronis + configures or shuts down. + """ if self._async_loop is None or self._async_loop.is_closed(): with self._async_lock: if self._async_loop is None or self._async_loop.is_closed(): self._async_loop = asyncio.new_event_loop() + + self._async_io_executor = ThreadPoolExecutor( + max_workers=self._async_io_workers, + thread_name_prefix="chronis-async-io-", + ) + self._async_loop.set_default_executor(self._async_io_executor) + self._async_thread = threading.Thread( target=self._async_loop.run_forever, daemon=True, @@ -139,4 +157,8 @@ async def _drain() -> None: self._async_loop = None self._async_thread = None + if self._async_io_executor is not None: + self._async_io_executor.shutdown(wait=wait) + self._async_io_executor = None + return drained diff --git a/chronis/core/execution/worker_pool.py b/chronis/core/execution/worker_pool.py new file mode 100644 index 0000000..8cbd726 --- /dev/null +++ b/chronis/core/execution/worker_pool.py @@ -0,0 +1,178 @@ +"""Bounded execution capacity for jobs, counted in permits rather than threads.""" + +import threading +from collections.abc import Callable +from typing import Any + +from chronis.utils.logging import ContextLogger + + +class Slot: + """One unit of the pool's capacity, held by exactly one job execution. + + Released when the job finishes — or abandoned, if the job timed out and its thread had + to be left behind. Both are idempotent: an abandoned thread that does eventually return + will try to release the slot a second time, and must not hand back capacity twice. + """ + + def __init__(self, pool: "WorkerPool") -> None: + self._pool = pool + self._lock = threading.Lock() + self._settled = False + self._thread: threading.Thread | None = None + + def run(self, func: Callable[[], Any], *, name: str) -> None: + """Run func on a daemon thread, releasing this slot when it returns.""" + + def _body() -> None: + try: + func() + finally: + self.release() + + thread = threading.Thread(target=_body, daemon=True, name=name) + self._thread = thread + self._pool._register(thread) + thread.start() + + def release(self) -> None: + """Hand this unit of capacity back. Safe to call more than once.""" + if self._settle(): + self._pool._release() + + def abandon(self) -> None: + """Hand the capacity back, but count the thread as leaked. + + Used when a job times out: Python cannot kill the thread running it, so the thread + is left to finish on its own — possibly never. The *capacity* must come back anyway, + or enough timeouts would leave the scheduler unable to run anything at all. + + The thread also stops being something shutdown waits for. Joining a thread we have + given up on would hang shutdown for exactly as long as the job hangs. + """ + if self._settle(): + if self._thread is not None: + self._pool._forget(self._thread) + self._pool._release(abandoned=True) + + def _settle(self) -> bool: + """Claim the right to settle this slot. True for the first caller only.""" + with self._lock: + if self._settled: + return False + self._settled = True + return True + + +class WorkerPool: + """Caps how many jobs run at once — sync and async alike. + + A ThreadPoolExecutor cannot do this job. A job killed by a timeout keeps its worker + thread forever (Python has no safe way to stop a thread) and the pool has no way to + replace that worker, so ``max_workers`` hung jobs permanently starve the node: it stops + executing anything, silently. And async jobs never touched the pool at all, so + ``max_workers`` never bounded them. + + So capacity is counted in *permits*, not threads. A job takes one permit whether it runs + on a thread or on the event loop, and a timed-out job returns its permit immediately + even though its thread lives on. The stranded threads are counted and reported, so the + leak is visible instead of presenting as a mysterious stall. + """ + + def __init__(self, max_workers: int, logger: ContextLogger | None = None) -> None: + self.max_workers = max_workers + self.logger = logger + + self._permits = threading.BoundedSemaphore(max_workers) + self._lock = threading.Lock() + self._in_use = 0 + self._abandoned = 0 + self._threads: list[threading.Thread] = [] + self._shutdown = False + + def reserve(self) -> Slot | None: + """Take one unit of capacity, or None if the pool is full. + + A caller that gets None must not claim the job: leaving it SCHEDULED lets the next + poll — or another node with spare capacity — pick it up. + """ + if self._shutdown: + return None + + if not self._permits.acquire(blocking=False): + return None + + with self._lock: + self._in_use += 1 + + return Slot(self) + + def stats(self) -> dict[str, int]: + """Capacity accounting, including threads stranded by timeouts.""" + with self._lock: + in_use = self._in_use + abandoned = self._abandoned + + return { + "max_workers": self.max_workers, + "in_use": in_use, + "available": self.max_workers - in_use, + "abandoned_threads": abandoned, + } + + def shutdown(self, wait: bool = True, timeout: float = 30.0) -> bool: + """Stop accepting work and wait for live jobs to finish. + + Threads abandoned by a timeout are not waited on — that is the whole point of + abandoning them, and joining one would hang shutdown for as long as the job hangs. + + Returns: + True if every live thread finished within the timeout. + """ + self._shutdown = True + + if not wait: + return False + + with self._lock: + threads = list(self._threads) + + deadline_all_done = True + for thread in threads: + thread.join(timeout=timeout) + if thread.is_alive(): + deadline_all_done = False + + return deadline_all_done + + def _register(self, thread: threading.Thread) -> None: + with self._lock: + self._threads = [t for t in self._threads if t.is_alive()] + self._threads.append(thread) + + def _forget(self, thread: threading.Thread) -> None: + """Stop waiting on a thread we have abandoned.""" + with self._lock: + self._threads = [t for t in self._threads if t is not thread] + + def _release(self, abandoned: bool = False) -> None: + with self._lock: + self._in_use -= 1 + if abandoned: + self._abandoned += 1 + abandoned_total = self._abandoned + else: + abandoned_total = 0 + + self._permits.release() + + if abandoned and self.logger: + self.logger.warning( + "Abandoned a timed-out job's thread; its capacity was returned but the " + "thread cannot be killed and may run indefinitely", + abandoned_threads=abandoned_total, + max_workers=self.max_workers, + ) + + +__all__ = ["WorkerPool", "Slot"] diff --git a/chronis/core/schedulers/fluent_builders.py b/chronis/core/schedulers/fluent_builders.py index f4a34da..995b59a 100644 --- a/chronis/core/schedulers/fluent_builders.py +++ b/chronis/core/schedulers/fluent_builders.py @@ -225,7 +225,13 @@ def config( timezone: Timezone for scheduling retry: Maximum retry attempts retry_delay: Delay between retries in seconds - timeout: Execution timeout in seconds + timeout: Execution timeout in seconds. What a timeout *does* depends on the + function: an async job is genuinely cancelled, while a sync job cannot be + — Python has no safe way to stop a thread — so it is marked failed and its + thread is left running. Either way the job raises JobTimeoutError and takes + the normal retry / on_failure path. Abandoned threads are counted by + scheduler.get_worker_status()['abandoned_threads']; a number that keeps + climbing means jobs are hanging well past their timeout. priority: Job priority (0-10, higher = more urgent) metadata: Custom metadata dict on_success: Success callback diff --git a/chronis/core/schedulers/polling_scheduler.py b/chronis/core/schedulers/polling_scheduler.py index 5277770..d91e0c9 100644 --- a/chronis/core/schedulers/polling_scheduler.py +++ b/chronis/core/schedulers/polling_scheduler.py @@ -236,9 +236,9 @@ def stop(self) -> dict[str, Any]: Returns: Dictionary with shutdown status: - - sync_jobs_completed: Whether the worker pool drained. A job abandoned by a - timeout may still be running — its thread cannot be killed — so this reports - that the pool shut down, not that every thread is gone. + - sync_jobs_completed: Whether every live worker thread finished. Threads + abandoned by a timeout are not waited on — they cannot be killed and may run + forever — so this says nothing about them. get_worker_status() counts them. - async_jobs_completed: Whether every in-flight async job finished. False if the 60s drain timed out and jobs were abandoned. - was_running: Whether scheduler was running before stop @@ -256,17 +256,32 @@ def stop(self) -> dict[str, Any]: self._runner.shutdown(wait=True) async_drained = self._execution_coordinator.shutdown_async(wait=True) - self._executor.shutdown(wait=True, cancel_futures=False) + sync_drained = self._workers.shutdown(wait=True) self._executor_shutdown = True self._running = False return { - "sync_jobs_completed": True, + "sync_jobs_completed": sync_drained, "async_jobs_completed": async_drained, "was_running": True, } + def get_worker_status(self) -> dict[str, int]: + """ + Execution capacity, for monitoring. + + Returns: + - max_workers: The configured cap on jobs running at once (sync and async) + - in_use: Jobs currently holding capacity + - available: Capacity left + - abandoned_threads: Threads left behind by timed-out sync jobs. Their capacity + was returned, but the threads themselves cannot be killed and may still be + running. A number that keeps climbing means jobs are hanging past their + timeout — the leak is in the job, not the scheduler. + """ + return self._workers.stats() + def get_queue_status(self) -> dict[str, Any]: """ Get job queue status for monitoring. @@ -821,7 +836,13 @@ def config( timezone: Timezone for scheduling retry: Maximum retry attempts retry_delay: Delay between retries in seconds - timeout: Execution timeout in seconds + timeout: Execution timeout in seconds. What a timeout *does* depends on the + function: an async job is genuinely cancelled, while a sync job cannot be + — Python has no safe way to stop a thread — so it is marked failed and its + thread is left running. Either way the job raises JobTimeoutError and takes + the normal retry / on_failure path. Abandoned threads are counted by + scheduler.get_worker_status()['abandoned_threads']; a number that keeps + climbing means jobs are hanging well past their timeout. priority: Job priority (0-10, higher = more urgent) metadata: Custom metadata dict diff --git a/docs/ARCHITECTURE_NOTES.md b/docs/ARCHITECTURE_NOTES.md index ba6ebe0..10611b1 100644 --- a/docs/ARCHITECTURE_NOTES.md +++ b/docs/ARCHITECTURE_NOTES.md @@ -33,30 +33,34 @@ to quietly cause duplicate execution. This makes `updated_at` core's field. Adapters must store it verbatim and never stamp their own clock on it — see `chronis.testing`, which checks this. +## How execution capacity is counted + +**`max_workers` caps jobs, not threads** — sync and async alike. Capacity is counted in +*permits* (`WorkerPool`), and a job takes one whether it runs on a worker thread or on the +event loop. A job with no permit available is simply not claimed; it stays SCHEDULED for the +next poll, or for a node with room. + +The reason capacity is not just "a thread in a pool" is the timeout. A sync job that times +out cannot be killed — Python has no safe way to stop a thread — so it is abandoned. Held on +a `ThreadPoolExecutor`, that abandoned thread would occupy a worker forever and the pool has +no way to replace it: `max_workers` hung jobs and the node silently stops executing anything +at all. Separating the permit from the thread means the *capacity* comes back immediately +even though the thread does not. + +What that costs, honestly: the OS thread is still leaked. It is counted +(`scheduler.get_worker_status()["abandoned_threads"]`) and logged, so it shows up as a +number that climbs rather than as a scheduler that mysteriously stalls — but a job that +hangs forever still holds a thread forever. The fix for that lives in the job, not here. + +Timeouts therefore mean different things by function colour, and this is documented on +`timeout` / `timeout_seconds`: an async job is genuinely cancelled (`asyncio.wait_for`), a +sync job is marked failed and abandoned. Both raise `JobTimeoutError` and take the same +retry / `on_failure` path. + --- ## Known weaknesses (not yet addressed) -### The concurrency model is four models - -There are, concurrently: a daemon thread per periodic task (`_PeriodicRunner`), a -`ThreadPoolExecutor` for sync jobs, a dedicated event-loop thread for async jobs, a -`threading.Timer` per timed sync job, and the event loop's own default executor for -post-execution storage writes. Consequences we know about and have not fixed: - -- **A timed-out sync job leaks a worker.** Python cannot safely kill a thread, so a timeout - marks the job failed and abandons the thread — which keeps occupying a pool slot until it - returns, possibly never. (We do release its *queue* slot immediately, so the job is not - also blocked from being re-enqueued. The pool slot is still gone.) Enough hung jobs will - starve the pool. -- **`max_workers` does not bound async jobs.** They run on the event loop, not the pool. For - an async-heavy workload the only bound is `max_queue_size`. -- **Timeout semantics differ by function colour.** An async job's timeout is a real - cancellation (`asyncio.wait_for`). A sync job's is mark-and-abandon. - -An asyncio-native core would collapse most of this. That is a large change and it is not -scheduled. - ### Misfire policy lives in three places Classification is in `core/misfire.py`, the SKIP branch in `PollingScheduler._get_ready_jobs` diff --git a/tests/integration/test_execution_capacity.py b/tests/integration/test_execution_capacity.py new file mode 100644 index 0000000..12bd80d --- /dev/null +++ b/tests/integration/test_execution_capacity.py @@ -0,0 +1,149 @@ +"""max_workers must be a real bound, and a hung job must not take the scheduler down with it.""" + +import asyncio +import threading +import time + +import pytest + +from chronis import InMemoryStorage, PollingScheduler +from chronis.core.state import JobStatus +from chronis.utils.time import utc_now + + +def _due_now(scheduler, job_id): + scheduler.storage.update_job(job_id, {"next_run_time": utc_now().isoformat()}) + scheduler._job_queue.add_job(job_id) + + +@pytest.mark.integration +class TestTimedOutJobsDoNotStarveTheScheduler: + """A timed-out sync job's thread cannot be killed, so it is abandoned. + + Its *capacity* must come back regardless. Held on a ThreadPoolExecutor, max_workers + hung jobs would permanently consume every worker and the node would stop executing + anything at all — silently. + """ + + def test_a_full_pool_of_hung_jobs_still_runs_the_next_job(self): + storage = InMemoryStorage() + scheduler = PollingScheduler( + storage_adapter=storage, polling_interval_seconds=1, max_workers=2 + ) + + hang = threading.Event() + ran = [] + scheduler.register_job_function("hang", lambda **kwargs: hang.wait(30)) + scheduler.register_job_function("normal", lambda **kwargs: ran.append(1)) + + # fill every worker with a job that will hang past its timeout + for i in range(2): + job = scheduler.create_interval_job( + func="hang", job_id=f"hang-{i}", seconds=300, timeout_seconds=1 + ) + _due_now(scheduler, job.job_id) + scheduler._execute_queued_jobs() + + time.sleep(2.5) # both time out; their threads are still stuck in wait() + + status = scheduler.get_worker_status() + assert status["in_use"] == 0 # capacity handed back + assert status["abandoned_threads"] == 2 # ...and the leak is visible, not silent + + scheduler.create_interval_job(func="normal", job_id="normal-1", seconds=300) + _due_now(scheduler, "normal-1") + scheduler._execute_queued_jobs() + time.sleep(1.5) + + assert ran == [1] # the scheduler is still alive + + hang.set() + scheduler.stop() + + def test_timed_out_job_is_still_marked_failed(self): + """Handing the capacity back must not skip the job's own outcome.""" + storage = InMemoryStorage() + scheduler = PollingScheduler( + storage_adapter=storage, polling_interval_seconds=1, max_workers=2 + ) + + hang = threading.Event() + scheduler.register_job_function("hang", lambda **kwargs: hang.wait(30)) + + job = scheduler.create_interval_job(func="hang", seconds=300, timeout_seconds=1) + _due_now(scheduler, job.job_id) + scheduler._execute_queued_jobs() + time.sleep(2.5) + + assert scheduler.get_job(job.job_id).status == JobStatus.FAILED + + hang.set() + scheduler.stop() + + +@pytest.mark.integration +class TestMaxWorkersBoundsAsyncJobs: + """Async jobs run on the event loop, never touching the worker threads. + + They used to be bounded by nothing: max_workers said one thing and the scheduler did + another. Capacity is now counted in permits, which async jobs hold too. + """ + + def test_async_jobs_respect_max_workers(self): + storage = InMemoryStorage() + scheduler = PollingScheduler( + storage_adapter=storage, polling_interval_seconds=1, max_workers=2 + ) + + running = [] + peak = [] + counter_lock = threading.Lock() + + async def slow(**kwargs): + with counter_lock: + running.append(1) + peak.append(len(running)) + await asyncio.sleep(0.4) + with counter_lock: + running.pop() + + scheduler.register_job_function("slow", slow) + + for i in range(8): + job = scheduler.create_interval_job(func="slow", job_id=f"async-{i}", seconds=300) + _due_now(scheduler, job.job_id) + + scheduler._execute_queued_jobs() + time.sleep(1.5) + + assert max(peak) <= 2 # max_workers, not "however many were queued" + + scheduler.stop() + + def test_saturated_pool_leaves_the_job_scheduled_for_the_next_poll(self): + """A job we have no room for must not be claimed — it would strand in RUNNING.""" + storage = InMemoryStorage() + scheduler = PollingScheduler( + storage_adapter=storage, polling_interval_seconds=1, max_workers=1 + ) + + hang = threading.Event() + scheduler.register_job_function("hang", lambda **kwargs: hang.wait(30)) + + scheduler.create_interval_job(func="hang", job_id="busy", seconds=300) + _due_now(scheduler, "busy") + scheduler._execute_queued_jobs() + time.sleep(0.3) + + assert scheduler.get_worker_status()["available"] == 0 + + scheduler.create_interval_job(func="hang", job_id="waiting", seconds=300) + _due_now(scheduler, "waiting") + scheduler._execute_queued_jobs() + + # Not claimed: still SCHEDULED, so the next poll (or another node) can take it + assert scheduler.get_job("waiting").status == JobStatus.SCHEDULED + assert scheduler.get_job("busy").status == JobStatus.RUNNING + + hang.set() + scheduler.stop() diff --git a/tests/unit/core/test_execute_queued_jobs.py b/tests/unit/core/test_execute_queued_jobs.py index 5695e89..38777d3 100644 --- a/tests/unit/core/test_execute_queued_jobs.py +++ b/tests/unit/core/test_execute_queued_jobs.py @@ -190,7 +190,7 @@ def test_lock_release_error_does_not_abort_batch(self): scheduler.lock.release = Mock(side_effect=ConnectionError("lock backend down")) scheduler._execute_queued_jobs() - scheduler._executor.shutdown(wait=True) + scheduler._workers.shutdown(wait=True) assert len(executed) == 3 assert scheduler._job_queue.get_available_slots() == 5 diff --git a/tests/unit/core/test_worker_pool.py b/tests/unit/core/test_worker_pool.py new file mode 100644 index 0000000..83bfb5d --- /dev/null +++ b/tests/unit/core/test_worker_pool.py @@ -0,0 +1,149 @@ +"""Unit tests for WorkerPool — capacity accounting that survives abandoned threads.""" + +import threading +import time + +from chronis.core.execution.worker_pool import WorkerPool + + +class TestReserve: + def test_reserve_yields_slots_up_to_max_workers(self): + pool = WorkerPool(max_workers=2) + + assert pool.reserve() is not None + assert pool.reserve() is not None + assert pool.reserve() is None # saturated + + def test_stats_track_usage(self): + pool = WorkerPool(max_workers=3) + pool.reserve() + + assert pool.stats() == { + "max_workers": 3, + "in_use": 1, + "available": 2, + "abandoned_threads": 0, + } + + def test_reserve_returns_none_after_shutdown(self): + pool = WorkerPool(max_workers=2) + pool.shutdown(wait=False) + + assert pool.reserve() is None + + +class TestRelease: + def test_release_returns_capacity(self): + pool = WorkerPool(max_workers=1) + slot = pool.reserve() + assert pool.reserve() is None + + slot.release() + + assert pool.reserve() is not None + + def test_release_is_idempotent(self): + """A thread abandoned by a timeout may return later and release a second time. + + If that gave the capacity back twice, the pool would slowly inflate past + max_workers and stop bounding anything. + """ + pool = WorkerPool(max_workers=2) + slot = pool.reserve() + + slot.release() + slot.release() + slot.release() + + assert pool.stats()["in_use"] == 0 + assert pool.reserve() is not None + assert pool.reserve() is not None + assert pool.reserve() is None # still capped at 2, not inflated + + +class TestAbandon: + def test_abandon_returns_capacity_but_counts_the_thread(self): + pool = WorkerPool(max_workers=1) + slot = pool.reserve() + + slot.abandon() + + assert pool.stats()["abandoned_threads"] == 1 + assert pool.reserve() is not None # capacity came back + + def test_abandon_is_idempotent(self): + pool = WorkerPool(max_workers=2) + slot = pool.reserve() + + slot.abandon() + slot.abandon() + + assert pool.stats()["abandoned_threads"] == 1 + assert pool.stats()["in_use"] == 0 + + def test_release_after_abandon_does_not_double_count(self): + """The abandoned thread finishes eventually and releases its slot on the way out.""" + pool = WorkerPool(max_workers=2) + slot = pool.reserve() + + slot.abandon() + slot.release() # the late-returning thread + + assert pool.stats()["in_use"] == 0 + assert pool.stats()["abandoned_threads"] == 1 + + +class TestRun: + def test_run_releases_the_slot_when_the_thread_returns(self): + pool = WorkerPool(max_workers=1) + slot = pool.reserve() + done = threading.Event() + + slot.run(done.set, name="job") + done.wait(2) + time.sleep(0.05) # let the thread unwind + + assert pool.stats()["in_use"] == 0 + + def test_run_releases_the_slot_even_when_the_job_raises(self): + pool = WorkerPool(max_workers=1) + slot = pool.reserve() + + def boom(): + raise RuntimeError("job exploded") + + slot.run(boom, name="job") + time.sleep(0.1) + + assert pool.stats()["in_use"] == 0 + assert pool.reserve() is not None + + +class TestShutdown: + def test_shutdown_waits_for_live_threads(self): + pool = WorkerPool(max_workers=2) + finished = [] + + slot = pool.reserve() + slot.run(lambda: (time.sleep(0.2), finished.append(1)), name="slow") + + assert pool.shutdown(wait=True, timeout=5) is True + assert finished == [1] + + def test_shutdown_does_not_wait_for_abandoned_threads(self): + """Joining a thread we gave up on would hang shutdown for as long as the job hangs.""" + pool = WorkerPool(max_workers=2) + hang = threading.Event() + + slot = pool.reserve() + slot.run(lambda: hang.wait(30), name="zombie") + time.sleep(0.05) + slot.abandon() + + started = time.time() + all_done = pool.shutdown(wait=True, timeout=5) + elapsed = time.time() - started + + assert elapsed < 1.0 # did not wait on the zombie + assert all_done is True + hang.set() diff --git a/tests/unit/services/test_execution_coordinator.py b/tests/unit/services/test_execution_coordinator.py index 347e480..83ae3bd 100644 --- a/tests/unit/services/test_execution_coordinator.py +++ b/tests/unit/services/test_execution_coordinator.py @@ -9,6 +9,7 @@ from chronis.core.execution.coordinator import ClaimedJob, ExecutionCoordinator, _OutcomeGate from chronis.core.execution.job_executor import JobExecutor +from chronis.core.execution.worker_pool import Slot, WorkerPool from chronis.core.state import JobStatus @@ -17,6 +18,18 @@ def claimed(job_data, original_scheduled_time=None): return ClaimedJob(data=job_data, original_scheduled_time=original_scheduled_time) +def a_slot(pool=None): + """Reserve real capacity, as try_execute does before handing a job to _trigger_execution.""" + slot = (pool or WorkerPool(max_workers=4)).reserve() + assert slot is not None + return slot + + +def mock_slot(): + """A stand-in slot, so a test can see whether the job was handed to a worker thread.""" + return Mock(spec=Slot) + + class TestTriggerExecutionLogging: """Test _trigger_execution logging behavior.""" @@ -25,7 +38,7 @@ def test_trigger_execution_does_not_log_info(self): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={}, @@ -36,23 +49,22 @@ def test_trigger_execution_does_not_log_info(self): ) coordinator._update_job_status = Mock() - coordinator.executor.submit = Mock(return_value=Mock(add_done_callback=Mock())) job_logger = Mock() - coordinator._trigger_execution(claimed({"job_id": "test-1"}), job_logger, Mock()) + coordinator._trigger_execution(claimed({"job_id": "test-1"}), job_logger, Mock(), a_slot()) # Verify info was NOT called job_logger.info.assert_not_called() def test_exception_during_trigger_is_logged(self): """Test that exception during execution trigger is logged.""" - executor_mock = Mock() - executor_mock.submit.side_effect = Exception("Submit failed") + failing_slot = mock_slot() + failing_slot.run.side_effect = Exception("Submit failed") coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=executor_mock, + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={}, @@ -65,7 +77,9 @@ def test_exception_during_trigger_is_logged(self): # Should raise exception and rollback with pytest.raises(Exception, match="Submit failed"): - coordinator._trigger_execution(claimed({"job_id": "test-1"}), job_logger, Mock()) + coordinator._trigger_execution( + claimed({"job_id": "test-1"}), job_logger, Mock(), failing_slot + ) # Verify warning was logged (rollback message) job_logger.warning.assert_called_once() @@ -85,7 +99,7 @@ def test_successful_execution_with_previous_retries_resets_count(self, mock_utc_ coordinator = ExecutionCoordinator( storage=storage_mock, lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={"test_func": lambda: None}, failure_handler_registry={}, success_handler_registry={}, @@ -145,7 +159,7 @@ def test_storage_error_during_retry_scheduling_is_logged(self, mock_utc_now, moc coordinator = ExecutionCoordinator( storage=storage_mock, lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={}, @@ -183,7 +197,7 @@ def failing_callback(job_id, job_info): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={"test-1": failing_callback}, @@ -224,7 +238,7 @@ def failing_global(job_id, job_info): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={}, @@ -260,7 +274,7 @@ def failing_callback(job_id, error, job_info): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={"test-1": failing_callback}, success_handler_registry={}, @@ -299,7 +313,7 @@ def failing_global(job_id, error, job_info): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={}, @@ -330,19 +344,17 @@ def failing_global(job_id, error, job_info): class TestExecutorSubmitRollback: """Test executor submit failure rollback behavior.""" - def test_trigger_execution_rolls_back_on_submit_failure(self): - """ThreadPool submit 실패 시 SCHEDULED로 복구.""" - # Mock storage + def test_trigger_execution_rolls_back_on_start_failure(self): + """job을 시작조차 못 하면 SCHEDULED로 복구.""" storage_mock = Mock() - # Mock executor that raises on submit - executor_mock = Mock() - executor_mock.submit.side_effect = RuntimeError("ThreadPool is shutting down") + failing_slot = mock_slot() + failing_slot.run.side_effect = RuntimeError("ThreadPool is shutting down") coordinator = ExecutionCoordinator( storage=storage_mock, lock=Mock(), - executor=executor_mock, + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={}, @@ -360,7 +372,7 @@ def test_trigger_execution_rolls_back_on_submit_failure(self): # Should raise RuntimeError with pytest.raises(RuntimeError, match="ThreadPool is shutting down"): - coordinator._trigger_execution(claimed(job_data), job_logger, Mock()) + coordinator._trigger_execution(claimed(job_data), job_logger, Mock(), failing_slot) # Verify status update - only one call for rollback # (RUNNING update now happens in try_execute before calling _trigger_execution) @@ -382,14 +394,11 @@ def test_trigger_execution_rolls_back_on_submit_failure(self): def test_trigger_execution_succeeds_normally(self): """정상적인 경우 rollback 발생 안 함.""" storage_mock = Mock() - executor_mock = Mock() - future_mock = Mock() - executor_mock.submit.return_value = future_mock coordinator = ExecutionCoordinator( storage=storage_mock, lock=Mock(), - executor=executor_mock, + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={}, @@ -401,20 +410,16 @@ def test_trigger_execution_succeeds_normally(self): job_data = {"job_id": "test-1", "status": "scheduled"} job_logger = Mock() + slot = mock_slot() # Should not raise - coordinator._trigger_execution(claimed(job_data), job_logger, Mock()) + coordinator._trigger_execution(claimed(job_data), job_logger, Mock(), slot) + slot.run.assert_called_once() # handed to a worker thread # RUNNING update는 이제 try_execute에서 발생하므로, _trigger_execution에서는 없어야 함 assert storage_mock.update_job.call_count == 0 - # Verify submit was called - executor_mock.submit.assert_called_once() - - # Verify callback registered - future_mock.add_done_callback.assert_called_once() - - # Verify no warning logged + # Verify no warning logged (i.e. no rollback) job_logger.warning.assert_not_called() @@ -430,7 +435,7 @@ async def async_func(): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={"async_func": async_func}, failure_handler_registry={}, success_handler_registry={}, @@ -450,11 +455,10 @@ async def async_func(): coordinator._job_executor._async_thread = Mock() with patch("asyncio.run_coroutine_threadsafe", return_value=mock_future) as mock_rcts: - coordinator._trigger_execution(claimed(job_data), Mock(), on_complete) + coordinator._trigger_execution(claimed(job_data), Mock(), on_complete, a_slot()) - # Async path should be used, not executor.submit + # Async path should be used; no worker thread is spawned for it mock_rcts.assert_called_once() - coordinator.executor.submit.assert_not_called() mock_future.add_done_callback.assert_called_once() def test_sync_job_dispatches_to_thread_pool(self): @@ -463,14 +467,12 @@ def test_sync_job_dispatches_to_thread_pool(self): def sync_func(): pass - executor_mock = Mock() - future_mock = Mock() - executor_mock.submit.return_value = future_mock + slot = mock_slot() coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=executor_mock, + workers=WorkerPool(max_workers=4), function_registry={"sync_func": sync_func}, failure_handler_registry={}, success_handler_registry={}, @@ -482,10 +484,10 @@ def sync_func(): job_data = {"job_id": "test-1", "func_name": "sync_func"} with patch("asyncio.run_coroutine_threadsafe") as mock_rcts: - coordinator._trigger_execution(claimed(job_data), Mock(), Mock()) + coordinator._trigger_execution(claimed(job_data), Mock(), Mock(), slot) - executor_mock.submit.assert_called_once() - mock_rcts.assert_not_called() + slot.run.assert_called_once() # handed to a worker thread + mock_rcts.assert_not_called() # not to the event loop class TestEnsureAsyncLoopThreadSafety: @@ -526,13 +528,13 @@ def test_rollback_failure_does_not_mask_original_error(self): storage_mock = Mock() storage_mock.update_job.side_effect = RuntimeError("Storage down") - executor_mock = Mock() - executor_mock.submit.side_effect = RuntimeError("ThreadPool is shutting down") + failing_slot = mock_slot() + failing_slot.run.side_effect = RuntimeError("ThreadPool is shutting down") coordinator = ExecutionCoordinator( storage=storage_mock, lock=Mock(), - executor=executor_mock, + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={}, @@ -545,7 +547,9 @@ def test_rollback_failure_does_not_mask_original_error(self): # Original error should propagate, not the rollback error with pytest.raises(RuntimeError, match="ThreadPool is shutting down"): - coordinator._trigger_execution(claimed({"job_id": "test-1"}), job_logger, Mock()) + coordinator._trigger_execution( + claimed({"job_id": "test-1"}), job_logger, Mock(), failing_slot + ) # Rollback was attempted storage_mock.update_job.assert_called_once() @@ -590,7 +594,7 @@ def _make_coordinator(self, **overrides): defaults = { "storage": Mock(), "lock": Mock(), - "executor": Mock(), + "workers": WorkerPool(max_workers=4), "function_registry": {}, "failure_handler_registry": {}, "success_handler_registry": {}, @@ -743,7 +747,7 @@ def global_callback(job_id, job_info): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={"test-1": job_callback}, @@ -784,7 +788,7 @@ def global_callback(job_id, error, job_info): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={"test-1": job_callback}, success_handler_registry={}, @@ -824,7 +828,7 @@ def callback(job_id, job_info): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={"test-1": callback}, @@ -866,7 +870,7 @@ def on_fail(job_id, error, job_info): coordinator = ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={"test_func": lambda: None}, failure_handler_registry={"test-1": on_fail}, success_handler_registry={}, @@ -905,7 +909,7 @@ def _make_coordinator(self): return ExecutionCoordinator( storage=Mock(), lock=Mock(), - executor=Mock(), + workers=WorkerPool(max_workers=4), function_registry={}, failure_handler_registry={}, success_handler_registry={}, From 16c81743a8859d445c1c4a268f9d9072dda0b1eb Mon Sep 17 00:00:00 2001 From: enginerd-kr Date: Sun, 12 Jul 2026 22:50:10 +0900 Subject: [PATCH 2/3] fix(core): keep a job's firing when it cannot be started MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found while measuring what happens as abandoned threads pile up: once thread creation starts failing, every job that fails to start silently skips the run it was claimed for. The claim has already advanced next_run_time to the *next* occurrence, so rolling back only the status leaves the job SCHEDULED with a run time in the future — invisible to the poller and to misfire detection, which only looks at run times in the past. An interval job with a 300s period simply does not run for another 300 seconds, with nothing to say it was dropped. Rollback now restores the scheduled time the claim consumed, which ClaimedJob has been carrying since the claim boundary was typed. The job stays due and the next poll runs it. Also records what abandoned threads actually cost, since the question has a sharp answer: a thread blocked on I/O is ~37 KB and releases the GIL (200 of them cost 1.25x on a Python workload — noise), while a thread spinning in Python holds the GIL and is ruinous (20 of them cost 17.6x). Reaching the OS thread limit is survivable: creation fails, the job rolls back intact, and the scheduler resumes when threads free up. Co-Authored-By: Claude Opus 4.8 (1M context) --- chronis/core/execution/coordinator.py | 32 +++++++++++---- docs/ARCHITECTURE_NOTES.md | 19 +++++++++ tests/integration/test_execution_capacity.py | 43 ++++++++++++++++++++ 3 files changed, 87 insertions(+), 7 deletions(-) diff --git a/chronis/core/execution/coordinator.py b/chronis/core/execution/coordinator.py index 6b24814..733db63 100644 --- a/chronis/core/execution/coordinator.py +++ b/chronis/core/execution/coordinator.py @@ -355,15 +355,33 @@ def _run() -> None: error=str(submit_error), error_type=type(submit_error).__name__, ) - try: - self._update_job_status(job_id, JobStatus.SCHEDULED) - except Exception as rollback_error: - job_logger.error( - "Failed to rollback job status after submit failure", - error=str(rollback_error), - ) + self._rollback_unstarted_claim(claimed, job_logger) raise submit_error + def _rollback_unstarted_claim(self, claimed: ClaimedJob, job_logger: ContextLogger) -> None: + """Undo a claim for a job that never actually started. + + The claim already moved next_run_time on to the *next* occurrence, so restoring only + the status would quietly drop the run we were about to perform: the job sits in + SCHEDULED with a future run time, invisible to both the poller and misfire detection, + and simply misses this firing. Put the scheduled time back so the job is still due + and the next poll picks it up. + """ + updates: JobUpdateData = { + "status": JobStatus.SCHEDULED.value, + "updated_at": canonical_utc_now(), + } + if claimed.original_scheduled_time: + updates["next_run_time"] = claimed.original_scheduled_time + + try: + self.storage.update_job(claimed.job_id, updates) + except Exception as rollback_error: + job_logger.error( + "Failed to rollback job after it could not be started", + error=str(rollback_error), + ) + def _execute_in_background( self, claimed: ClaimedJob, diff --git a/docs/ARCHITECTURE_NOTES.md b/docs/ARCHITECTURE_NOTES.md index 10611b1..7386bba 100644 --- a/docs/ARCHITECTURE_NOTES.md +++ b/docs/ARCHITECTURE_NOTES.md @@ -52,6 +52,25 @@ What that costs, honestly: the OS thread is still leaked. It is counted number that climbs rather than as a scheduler that mysteriously stalls — but a job that hangs forever still holds a thread forever. The fix for that lives in the job, not here. +### What abandoned threads actually cost + +It depends entirely on what the job is stuck on, and the two cases are not close (measured +on CPython 3.12, macOS): + +| Abandoned threads | Cost | +|---|---| +| Blocked on I/O, a lock, a DB call, `sleep` — the usual case | **~37 KB RSS each**, and no CPU: a blocked thread releases the GIL. 200 of them slowed a Python workload by 1.25x, i.e. noise. A thousand of them is ~37 MB. | +| Spinning in Python (a runaway loop) | **Ruinous.** They hold the GIL. *Twenty* of them slowed the same workload by **17.6x** — the whole scheduler crawls. No counter can save you from this; it only tells you where to look. | + +The terminal state is the OS thread limit (`ulimit -u`; 4000+ on a typical dev machine). +Reaching it does not crash anything: thread creation fails, the job is rolled back to +SCHEDULED with its scheduled time intact, its permit is returned, and the scheduler picks it +up again once threads are free. It simply cannot start new work in the meantime. + +So: leaking blocked threads is survivable and visible long before it is fatal. Leaking +spinning threads is not survivable, and the counter is the fastest way to find out which +job is doing it. + Timeouts therefore mean different things by function colour, and this is documented on `timeout` / `timeout_seconds`: an async job is genuinely cancelled (`asyncio.wait_for`), a sync job is marked failed and abandoned. Both raise `JobTimeoutError` and take the same diff --git a/tests/integration/test_execution_capacity.py b/tests/integration/test_execution_capacity.py index 12bd80d..3a11111 100644 --- a/tests/integration/test_execution_capacity.py +++ b/tests/integration/test_execution_capacity.py @@ -147,3 +147,46 @@ def test_saturated_pool_leaves_the_job_scheduled_for_the_next_poll(self): hang.set() scheduler.stop() + + +@pytest.mark.integration +class TestUnstartableJobKeepsItsFiring: + """A job that cannot be started must not lose the run it was claimed for. + + The claim has already moved next_run_time on to the next occurrence. Rolling back only + the status leaves the job SCHEDULED with a future run time — invisible to the poller and + to misfire detection — so the firing is silently dropped. Under thread exhaustion that + happens to every job that fails to start. + """ + + def test_rollback_restores_the_scheduled_time(self): + from unittest.mock import patch + + storage = InMemoryStorage() + scheduler = PollingScheduler( + storage_adapter=storage, polling_interval_seconds=1, max_workers=4 + ) + ran = [] + scheduler.register_job_function("f", lambda **kwargs: ran.append(1)) + + scheduler.create_interval_job(func="f", job_id="J", seconds=300) + _due_now(scheduler, "J") + was_due_at = storage.get_job("J")["next_run_time"] + + with patch.object( + threading.Thread, "start", side_effect=RuntimeError("can't start new thread") + ): + scheduler._execute_queued_jobs() + + job = storage.get_job("J") + assert job["status"] == JobStatus.SCHEDULED.value + assert job["next_run_time"] == was_due_at # still due; the run was not dropped + + # once threads are available again, the job runs — it did not skip to +300s + scheduler._job_queue.add_job("J") + scheduler._execute_queued_jobs() + time.sleep(0.5) + + assert ran == [1] + + scheduler.stop() From 785bd8db32b1116f945a9048e5ca717e71bea26e Mon Sep 17 00:00:00 2001 From: enginerd-kr Date: Sun, 12 Jul 2026 22:56:46 +0900 Subject: [PATCH 3/3] docs: record the execution models considered beyond threads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers "what would replace the thread model?" with measurements instead of instinct. The residual problems split into two axes: scheduler machinery (an asyncio-native core would collapse the loops and timers, but buys no safety) and job isolation (only a process makes timeout a real guarantee — terminate() kills even a CPU-spinning child in ~1ms, the case that costs the thread model 17.6x in GIL contention). The recorded path, when picked up: per-job opt-in process isolation on the permit model (config(isolation="process")). WorkerPool already decoupled capacity from threads, so a ProcessSlot is an extension rather than a rewrite. Constraints noted: 1.5-25ms dispatch per process, and function resolution needs a worker initializer (lambda registrations cannot cross a process boundary) — while args/kwargs already round-trip as JSON by the storage contract. Co-Authored-By: Claude Fable 5 --- docs/ARCHITECTURE_NOTES.md | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/docs/ARCHITECTURE_NOTES.md b/docs/ARCHITECTURE_NOTES.md index 7386bba..1e2afab 100644 --- a/docs/ARCHITECTURE_NOTES.md +++ b/docs/ARCHITECTURE_NOTES.md @@ -76,6 +76,40 @@ 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. +### Beyond threads: execution models considered + +The thread model's residual problems are really two independent axes, and the candidate +models each fix a different one: + +- **Axis A — scheduler machinery.** Three daemon loops plus a `threading.Timer` per timed + job is inelegant. An asyncio-native core would collapse all of it into one event loop and + make async cancellation first-class. But it does nothing for axis B — sync jobs still have + to run *somewhere* killable — so it is a large rewrite with no safety payoff. Long-term + track, not scheduled. + +- **Axis B — job isolation.** Only a process can make `timeout` a real guarantee. Measured + (CPython 3.12, macOS): `terminate()` kills a sleep-hung child in ~20 ms and a + **CPU-spinning one in ~1 ms** — the exact case that poisons the GIL in the thread model + (17.6x slowdown from twenty spinners). Dispatch costs 1.5 ms/job (fork) to 25 ms/job + (spawn) versus 0.03 ms for a thread — acceptable at Chronis's polling cadence. The real + constraint is function resolution: a child process has no registry, so names registered + against lambdas/closures cannot be resolved there; a worker-initializer hook is needed. + Notably, job `args`/`kwargs` already round-trip through storage as JSON, so the data side + of the serialization constraint is already met by contract. + + Subinterpreters (PEP 734) and free-threaded 3.13t would fix the GIL-poisoning but still + cannot kill a hung job, and their ecosystems are young — not candidates yet. + +The path we would take, when this is picked up: **per-job opt-in process isolation** on top +of the permit model — `config(isolation="process")`. `WorkerPool` already decoupled capacity +from threads, so a `ProcessSlot` beside the thread slot is an extension, not a rewrite. +Threads stay the default (cheap, API unchanged); jobs whose timeout must be a hard guarantee +opt into a process where timeout means SIGKILL — no zombie, no GIL poisoning. Celery's +pluggable pools reached the same conclusion. + +Deployment-level backstop regardless of model: alert on `abandoned_threads` (or wire it into +a liveness probe) so a node accumulating zombies gets recycled instead of limping. + --- ## Known weaknesses (not yet addressed)