Bound celery pool-child RSS so one runaway task can't take the container

src/ol_infrastructure/applications/mit_learn/__main__.py

@@ -1600,6 +1600,24 @@ def _resource_config(config_key: str, default: dict[str, str]) -> dict[str, str]
                 redis_password=redis_config.require("password"),
                 resource_requests=celery_default_resource_requests,
                 resource_limits=celery_default_resource_limits,
+                # 640Mi. A healthy child on this queue sits at ~165Mi (observed
+                # on applications-production 2026-08-17: whole container ~480Mi
+                # for master + 2 children), so this only fires on a child that
+                # has genuinely ballooned, not in steady state.
+                #
+                # Sized against the *floor*-derived limit (1Gi request x the 2:1
+                # ratio = 2Gi), not the 2560Mi declared above or the 6144Mi the
+                # VPA is currently enforcing, because the floor is the smallest
+                # limit a pod can run under: 150Mi master + 2 x (640 + one task's
+                # growth) has to clear 2Gi. Coupled to --concurrency=2 and to
+                # _worker_vpa_bounds["min_allowed"] below -- revisit all three
+                # together.
+                #
+                # This bounds *carry-over* only. A single task that allocates
+                # past the cgroup limit in one go still OOM-kills the container,
+                # because celery checks RSS between tasks. See the 0.77.3
+                # get_learning_resource_views regression.
+                max_memory_per_child_kib=655360,
             ),
             OLApplicationK8sCeleryWorkerConfig(
                 queue_name="edx_content",

src/ol_infrastructure/components/services/k8s.py

@@ -165,6 +165,16 @@ class OLApplicationK8sCeleryWorkerConfig(BaseModel):
     min_replicas: NonNegativeInt = 1
     max_replicas: NonNegativeInt = 10
     autoscale_queue_depth: NonNegativeInt = 10
+    # Resident set size, in KiB, above which celery retires a pool child. Checked
+    # after each task returns, so it bounds what a child *carries into the next
+    # task* -- it cannot stop a single task that blows the cgroup limit on its own.
+    # Without it the only recycle trigger is --max-tasks-per-child (100), so a
+    # child that balloons on task 1 stays resident for 99 more and the kernel
+    # OOM-kills the whole container, taking unrelated in-flight tasks with it.
+    # Size it so master + concurrency * (cap + one task's growth) clears the
+    # *smallest* limit the pod can run under, which under a VPA is the floor-
+    # derived limit, not the declared one.
+    max_memory_per_child_kib: PositiveInt | None = None
     redis_database_index: str = "1"
     redis_host: Output[str]
     redis_password: str
@@ -2077,6 +2087,18 @@ def _build_keda_triggers(
                                         celery_worker_config.log_level,
                                         "--max-tasks-per-child",  # Max number of tasks the pool worker will process before being replaced
                                         "100",
+                                        *(
+                                            [
+                                                # Max RSS (KiB) a pool worker may
+                                                # hold before being replaced
+                                                "--max-memory-per-child",
+                                                str(
+                                                    celery_worker_config.max_memory_per_child_kib
+                                                ),
+                                            ]
+                                            if celery_worker_config.max_memory_per_child_kib
+                                            else []
+                                        ),
                                         "--concurrency=2",  # Don't try to use all cores on node
                                         "--prefetch-multiplier=1",
                                     ],

tests/ol_infrastructure/components/services/test_k8s_extra_containers.py

@@ -970,3 +970,65 @@ def check(spec):
         }
 
     return app.webapp_pod_monitor.spec.apply(check)
+
+
+# ─── celery --max-memory-per-child ────────────────────────────────────────────
+
+
+def _celery_worker_config(**overrides) -> OLApplicationK8sCeleryWorkerConfig:
+    defaults = {
+        "application_name": "memcapped",
+        "worker_name": "default",
+        "redis_host": pulumi.Output.from_input("redis.example.com"),
+        "redis_password": "hunter2",  # pragma: allowlist secret
+    }
+    defaults.update(overrides)
+    return OLApplicationK8sCeleryWorkerConfig(**defaults)
+
+
+@pulumi.runtime.test
+def test_max_memory_per_child_omitted_by_default():
+    """Existing OLApplicationK8s consumers must be unaffected.
+
+    The flag changes when celery retires a pool child, so it has to stay opt-in
+    rather than arriving with a default that silently reshapes every other
+    application's worker recycling behaviour.
+    """
+    app = OLApplicationK8s(
+        _base_config(
+            application_name="memcapped",
+            celery_worker_configs=[_celery_worker_config()],
+        )
+    )
+
+    def check(containers):
+        worker = next(c for c in containers if c["name"] == "celery-worker")
+        assert "--max-memory-per-child" not in worker["command"]
+        # the sibling recycle trigger stays unconditional
+        assert "--max-tasks-per-child" in worker["command"]
+
+    return app.celery_deployments[0].spec.template.spec.containers.apply(check)
+
+
+@pulumi.runtime.test
+def test_max_memory_per_child_emitted_as_flag_and_value():
+    app = OLApplicationK8s(
+        _base_config(
+            application_name="memcapped",
+            celery_worker_configs=[
+                _celery_worker_config(max_memory_per_child_kib=655360)
+            ],
+        )
+    )
+
+    def check(containers):
+        command = next(c for c in containers if c["name"] == "celery-worker")["command"]
+        # celery takes the value as a separate argv entry, not --flag=value
+        assert command[command.index("--max-memory-per-child") + 1] == "655360"
+
+    return app.celery_deployments[0].spec.template.spec.containers.apply(check)
+
+
+def test_max_memory_per_child_rejects_non_positive():
+    with pytest.raises(ValidationError):
+        _celery_worker_config(max_memory_per_child_kib=0)