Merge pull request #1534 from PolicyEngine/add-modal-concurrent-inputs-and-max-cap

Set Modal worker concurrency to 5 and cap autoscale at 100

Author: hua7450

changelog.d/1533.changed.md

@@ -0,0 +1 @@
+Enable Modal `@modal.concurrent(max_inputs=5, target_inputs=4)` (5x warm-pool capacity, autoscaler aims for 80% steady-state utilisation, no new container cost) and cap autoscale at `max_containers=100` to prevent runaway scaling.

policyengine_household_api/modal_release/worker_app.py

@@ -46,6 +46,13 @@ def worker_function_options(
         "timeout": 180,
         "scaledown_window": 300,
         "enable_memory_snapshot": True,
+        # Hard cap on autoscale. Without this Modal is bounded only by the
+        # workspace quota, so a runaway traffic spike (or a buggy partner
+        # client) could scale to hundreds of containers and rack up cost.
+        # 100 covers any realistic partner burst we expect today (peak 100
+        # concurrent x 5 inputs = 500 in-flight) while keeping accidents
+        # bounded.
+        "max_containers": 100,
     }
     if environment == "main":
         options["min_containers"] = 3
@@ -54,7 +61,21 @@ def worker_function_options(
     return options
 
 
+def worker_concurrency_options() -> dict[str, int]:
+    # Each container processes up to 5 requests in parallel (`max_inputs`).
+    # With ~3s of CPU per request on a 1-core container, 5-way sharing gives
+    # ~15s wall-time per request when fully saturated. Multiplies effective
+    # warm-pool capacity 5x with no additional container cost.
+    #
+    # `target_inputs=4` is the autoscaler's steady-state goal: keep average
+    # utilisation at 80% so each container retains one free slot to absorb
+    # single-request spikes without waiting on a cold start. Containers still
+    # burst up to `max_inputs=5` under load before queueing.
+    return {"max_inputs": 5, "target_inputs": 4}
+
+
 @app.cls(**worker_function_options())
+@modal.concurrent(**worker_concurrency_options())
 class HouseholdWorker:
     """Worker class for handling household API requests.
 

tests/unit/modal_release/test_worker_app.py

@@ -84,6 +84,48 @@ def test_household_worker_exposes_post_snapshot_reset_hook(worker_app):
     assert hasattr(worker_cls, "reset_post_snapshot_state")
 
 
+def test_worker_concurrency_options_set_max_inputs(worker_app):
+    """Each container processes up to 5 requests in parallel. Keeping this
+    consistent across environments ensures staging behavior mirrors
+    production so concurrency-related issues surface in staging tests."""
+    assert worker_app.worker_concurrency_options()["max_inputs"] == 5
+
+
+def test_worker_concurrency_options_set_target_inputs(worker_app):
+    """`target_inputs=4` keeps the autoscaler aiming for 80% steady-state
+    utilisation, so each container retains a free slot to absorb a
+    single-request spike without waiting on a cold start."""
+    assert worker_app.worker_concurrency_options()["target_inputs"] == 4
+
+
+def test_worker_function_options_do_not_use_deprecated_concurrency_kwarg(
+    worker_app,
+):
+    for environment in ("main", "staging", "testing"):
+        options = worker_app.worker_function_options(
+            modal_environment=environment
+        )
+        assert "allow_concurrent_inputs" not in options, (
+            "`allow_concurrent_inputs` is deprecated; use "
+            f"`@modal.concurrent` for `{environment}` worker concurrency"
+        )
+
+
+def test_worker_function_options_max_containers_capped_in_all_envs(
+    worker_app,
+):
+    """A hard ceiling on autoscale prevents runaway scaling from a buggy
+    client or traffic spike from racking up unbounded cost."""
+    for environment in ("main", "staging", "testing"):
+        options = worker_app.worker_function_options(
+            modal_environment=environment
+        )
+        assert options["max_containers"] == 100, (
+            f"max_containers must be 100 in `{environment}` to bound "
+            "autoscale cost"
+        )
+
+
 def test_country_package_install_specs_use_release_package_versions_only():
     assert country_package_install_specs(
         {