From a6342683890bf357dd21ad3e1afc209b3f45e367 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Tue, 21 Jul 2026 10:19:10 +0300 Subject: [PATCH 1/3] fix: open root container for non-forking worker pools setup_di opened the root container only via worker_process_init, 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 under those pools the root container was never opened and every @inject task raised ContainerClosedError. Connect the same open/close handlers to worker_init/worker_shutdown, which fire once in the main worker process for all pools. The existing worker_process_init/worker_process_shutdown pair is unchanged, still giving each forked process (prefork) its own open/close cycle. --- architecture/dependency-injection.md | 45 ++++++++++++++++++---------- modern_di_celery/main.py | 13 ++++++++ 2 files changed, 43 insertions(+), 15 deletions(-) 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 From 893b92f66e93de2d010253e739c889a4de68bb21 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Tue, 21 Jul 2026 10:19:15 +0300 Subject: [PATCH 2/3] test: worker_init opens root without worker_process_init Simulates the gevent/eventlet/threads path: build a fresh app and container, fire worker_init directly (never worker_process_init), and assert the root container opens and an @inject task resolves. Fails without the fix, since only worker_process_init was connected before. --- tests/test_lifespan.py | 33 +++++++++++++++++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) 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 From 21a23d2f451069bb635a75370b8e809fa8f992d5 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Tue, 21 Jul 2026 10:19:20 +0300 Subject: [PATCH 3/3] docs(release): add 3.0.1 notes Release notes for the non-forking worker pool fix, plus the lightweight-lane planning change file recording the defect and fix. --- .../2026-07-21.01-open-root-all-pools.md | 49 +++++++++++++++++++ planning/releases/3.0.1.md | 28 +++++++++++ 2 files changed, 77 insertions(+) create mode 100644 planning/changes/2026-07-21.01-open-root-all-pools.md create mode 100644 planning/releases/3.0.1.md 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.