diff --git a/architecture/dependency-injection.md b/architecture/dependency-injection.md index cfd0f7e..72314fc 100644 --- a/architecture/dependency-injection.md +++ b/architecture/dependency-injection.md @@ -13,35 +13,50 @@ is `setup_di`, `fetch_di_container`, `FromDI`, `inject`, and `DITask`. (`"modern_di_container"`), a named constant — writer (`setup_di`) and reader (`fetch_di_container`) stay in provable agreement instead of relying on a bare string literal. -2. Connects two named closures to Celery's worker-process signals: - `signals.worker_process_init.connect(_open_container, weak=False)` calls - `container.open()`, and - `signals.worker_process_shutdown.connect(_close_container, weak=False)` - calls `container.close_sync()`. - -`weak=False` is required on both connections. Celery's signal dispatcher +2. Connects two named closures to **two pairs** of Celery worker signals: + `worker_process_init`/`worker_process_shutdown` and + `worker_init`/`worker_shutdown`. Both pairs call the same closures — + `_open_container` (`container.open()`) and `_close_container` + (`container.close_sync()`). + +The two pairs cover different pool families. `worker_process_init`/ +`worker_process_shutdown` fire only under the **prefork** and **solo** pools +(once per forked child for prefork), giving each forked process its own +open/close so cached resources and finalizers are fork-safe. `worker_init`/ +`worker_shutdown` fire once in the main worker process for **all** pools — +this is the *only* pair the **gevent**, **eventlet**, and **threads** pools +send, since those pools run tasks in the main process and never fork. Without +`worker_init`/`worker_shutdown`, the root container would never open under +those pools and every `@inject` task would raise `ContainerClosedError`. The +overlap between the pairs under prefork/solo is harmless: `Container.open()` +is a no-op when already open, and `close_sync()` is a no-op when nothing was +cached. + +`weak=False` is required on all four connections. Celery's signal dispatcher holds receivers by weak reference by default; `_open_container` and `_close_container` are local closures with no other strong reference keeping them alive, so a weak-ref connection would let the garbage collector reclaim -them before a forked worker process ever fires the signal — the handlers -would silently never run. `weak=False` makes the dispatcher hold a strong -reference for the life of the connection instead. +them before a worker process ever fires the signal — the handlers would +silently never run. `weak=False` makes the dispatcher hold a strong reference +for the life of each connection instead. `fetch_di_container(app)` reads the same key back off `app.conf` and returns the root container. ## Lifecycle -Reopening on `worker_process_init` is idempotent: a fresh `Container` is -already open on construction, and `Container.open` is a no-op when already -open, so firing `worker_process_init` again — a restart, a test re-entry — -reopens a container that was closed on a previous `worker_process_shutdown` +Reopening is idempotent: a fresh `Container` is already open on construction, +and `Container.open` is a no-op when already open, so firing an open signal +again — a restart, a test re-entry, or the second signal of an overlapping +pair — reopens a container that was closed on a previous shutdown signal instead of raising an error. This matters because Celery's prefork pool forks one worker process per configured concurrency slot, and each forked process fires its own `worker_process_init`/`worker_process_shutdown` pair independently on the container object it inherited from the fork — the open/close cycle is scoped per forked worker process, not to the parent -process as a whole. +process as a whole. Under gevent/eventlet/threads there is no fork at all; +`worker_init`/`worker_shutdown` scope the same open/close cycle to the single +main worker process instead. ## Per-task scope diff --git a/modern_di_celery/main.py b/modern_di_celery/main.py index 5f245ad..bdea2a1 100644 --- a/modern_di_celery/main.py +++ b/modern_di_celery/main.py @@ -23,8 +23,21 @@ def _close_container(**_: typing.Any) -> None: # noqa: ANN401 # weak=False is required: Celery signals default to weak refs, which would # garbage-collect these handlers before a worker process ever fires them. + # + # worker_process_init/worker_process_shutdown fire only under the prefork and + # solo pools (once per forked child for prefork), giving each forked process + # its own open/close so cached resources and finalizers are fork-safe. + # worker_init/worker_shutdown fire once in the main worker process for ALL + # pools, which is the only signal the gevent/eventlet/threads pools send — + # those pools never fork and never emit worker_process_init, so without this + # the root container would stay closed and every @inject task would raise + # ContainerClosedError. The overlap between the two pairs is harmless: + # Container.open() is a no-op when already open, and close_sync() is a no-op + # when nothing was cached. signals.worker_process_init.connect(_open_container, weak=False) signals.worker_process_shutdown.connect(_close_container, weak=False) + signals.worker_init.connect(_open_container, weak=False) + signals.worker_shutdown.connect(_close_container, weak=False) return container diff --git a/planning/changes/2026-07-21.01-open-root-all-pools.md b/planning/changes/2026-07-21.01-open-root-all-pools.md new file mode 100644 index 0000000..92b92bc --- /dev/null +++ b/planning/changes/2026-07-21.01-open-root-all-pools.md @@ -0,0 +1,49 @@ +--- +summary: `setup_di` now also opens/closes the root container on `worker_init`/`worker_shutdown`, so the gevent/eventlet/threads pools (which never fire `worker_process_init`) get a working root container instead of every `@inject` task raising `ContainerClosedError`. +--- + +# Change: Open the root container for non-forking worker pools + +**Lane:** lightweight — one file (~13 LOC net) + one new test, no new file, +no public-API change, single straightforward test. + +## Goal + +`setup_di` opened the root container only via `worker_process_init` (closed +via `worker_process_shutdown`). Celery sends `worker_process_init` only from +the **prefork** and **solo** pools. The **gevent**, **eventlet**, and +**threads** pools run tasks in the main worker process and never send it — so +under those pools the root container was never opened, and modern-di 3.0's +mandatory-open lifecycle made every `@inject` task raise `ContainerClosedError`. + +## Approach + +Connect the existing `_open_container`/`_close_container` closures to +`worker_init`/`worker_shutdown` *in addition to* the existing +`worker_process_init`/`worker_process_shutdown` — the per-process pair is +unchanged. `worker_init`/`worker_shutdown` fire once in the main worker +process for all pools, covering gevent/eventlet/threads. +`worker_process_init`/`worker_process_shutdown` still fire per forked child +(prefork) and in solo, giving each forked process its own open/close so +cached resources and finalizers stay fork-safe. The overlap under +prefork/solo is harmless: `Container.open()` is a no-op on an already-open +container, and `close_sync()` is a no-op when nothing was cached. + +## Files + +- `modern_di_celery/main.py` — `setup_di` connects `_open_container`/ + `_close_container` to `worker_init`/`worker_shutdown` as well. +- `tests/test_lifespan.py` — new test fires `worker_init`/`worker_shutdown` + directly (without `worker_process_init`) against a fresh app/container to + prove the new signals — not the test fixture — open/close it. +- `architecture/dependency-injection.md` — "Setup and worker-signal + lifecycle" and "Lifecycle" sections updated to describe both signal pairs. + +## Verification + +- [x] Failing test first: with the fix reverted, the new test fails at + `assert container.closed is False` after `signals.worker_init.send(...)`. +- [x] Apply the change. +- [x] Test passes with the fix restored. +- [x] `just test-ci` — full suite green, 100% line coverage. +- [x] `just lint-ci` — clean. diff --git a/planning/releases/3.0.1.md b/planning/releases/3.0.1.md new file mode 100644 index 0000000..21808eb --- /dev/null +++ b/planning/releases/3.0.1.md @@ -0,0 +1,28 @@ +# modern-di-celery 3.0.1 — non-forking pool fix + +A patch release fixing the root container never opening under the +gevent/eventlet/threads worker pools. + +## Fix + +- **The root container now opens for the non-forking gevent/eventlet/threads + pools too.** `setup_di` opened the root container only via + `worker_process_init` (closed via `worker_process_shutdown`), which Celery + sends only from the **prefork** and **solo** pools. The **gevent**, + **eventlet**, and **threads** pools run tasks in the main worker process + and never send it, so the root container was never opened under those + pools and every `@inject` task raised `ContainerClosedError`. `setup_di` + now also connects the same open/close handlers to `worker_init`/ + `worker_shutdown`, which fire once in the main worker process for all + pools. `worker_process_init`/`worker_process_shutdown` are unchanged — + prefork and solo still get their own per-forked-process open/close cycle. + +## Downstream + +No action needed. If you run Celery workers on prefork or solo, nothing +changes. If you run gevent, eventlet, or threads, `@inject`/`DITask` tasks +that previously raised `ContainerClosedError` on every call now work. + +## Internals + +- 100% line coverage; `ruff`, `ty` clean. diff --git a/tests/test_lifespan.py b/tests/test_lifespan.py index a300008..e3003ca 100644 --- a/tests/test_lifespan.py +++ b/tests/test_lifespan.py @@ -1,9 +1,11 @@ +import typing + from celery import Celery, signals from modern_di import Container import modern_di_celery -from modern_di_celery import fetch_di_container -from tests.dependencies import Dependencies +from modern_di_celery import FromDI, fetch_di_container, inject +from tests.dependencies import Dependencies, DependentCreator, SimpleCreator def test_fetch_returns_the_same_container(app: Celery) -> None: @@ -32,3 +34,30 @@ def test_restart_reopens_without_error(app: Celery) -> None: assert container.closed is True signals.worker_process_init.send(sender=None) # second cycle must not raise ContainerClosedError assert container.closed is False + + +def test_worker_init_opens_container_for_non_forking_pools() -> None: + # gevent/eventlet/threads run in the main worker process and never fire + # worker_process_init — only worker_init. Use a fresh app/container (not the + # `app` fixture, which auto-opens) so this proves the signal opens it. + non_forking_app = Celery("non-forking", broker="memory://", backend="cache+memory://") + non_forking_app.conf.task_always_eager = True + non_forking_app.conf.task_store_eager_result = True + container = modern_di_celery.setup_di(non_forking_app, Container(groups=[Dependencies], validate=True)) + assert container.closed is True + + signals.worker_init.send(sender=None) # no worker_process_init: simulates gevent/eventlet/threads + assert container.closed is False + + @non_forking_app.task + @inject + def sample( + app_instance: typing.Annotated[SimpleCreator, FromDI(SimpleCreator)], + request_instance: typing.Annotated[DependentCreator, FromDI(Dependencies.request_factory)], + ) -> bool: + return isinstance(app_instance, SimpleCreator) and isinstance(request_instance, DependentCreator) + + assert sample.delay().get() is True + + signals.worker_shutdown.send(sender=None) + assert container.closed is True