fix(dagster-aws): guarantee ECS pipes interruption subprocess is reaped (#25004)

## Summary & Motivation

On master build
[153444](https://buildkite.com/dagster/internal/builds/153444), the
`dagster-aws-3-12` step hung: every `pipes_tests` test from
`test_ecs_pipes_task_failed_to_start` onward took *exactly* 240.0s — the
global `pytest-timeout` fallback (`timeout = "240"`, `method: signal`)
killing each hung test and limping to the next. Every pipes test
*before* that point ran in seconds, which localizes the problem to
mid-module state corruption rather than a slow agent or network issue.

The corruption is introduced by
`test_ecs_pipes_interruption_forwarding`, which #24924 rewrote. Its
happy path was a single `p.terminate()` + `p.join(timeout=30)` + `assert
not p.is_alive()`, with `p.kill()` escalation present *only* in the
"task never launched within 60s" error branch. The subprocess under test
runs `materialize`, which installs Dagster's
SIGTERM→graceful-interruption handler — the whole point of the test.
When that graceful shutdown outlives the 30s join (or any of the
assertions fails), the happy path returns without ever calling
`p.kill()`, leaving the child **alive**.

That child is `fork()`ed from the test process and holds the inherited
moto server socket on `localhost:5193` (the shared, function-scoped
`moto_server` fixture port). A leaked orphan wedges that port, so every
subsequent moto-dependent pipes test blocks until the 240s timeout
fires. The build's 32.7s runtime for
`test_ecs_pipes_interruption_forwarding` (≈ startup + the full 30s join
cap) is consistent with the join expiring and the child surviving
SIGTERM.

## Changes

- **Always reap the child.** Wrap the test body in `try/finally` and
unconditionally `p.kill()` + `p.join()` the subprocess on every exit
path (assertion failure, join timeout, or the "didn't launch" branch),
so it can never be leaked.
- **Don't inherit the moto socket.** Start the subprocess from a `spawn`
multiprocessing context (`multiprocessing.get_context("spawn")`, used
for the `Manager`, `Event`, and `Process`). A spawned child gets a fresh
interpreter and does not inherit this process's file descriptors —
including the moto listening socket — and reaches moto over the network
as a client instead. Even a leaked child can then no longer wedge port
5193. Linux CI defaults to `fork`; macOS already defaults to `spawn`.
- **Gate on a real completion signal.** Add a
`materialization_done_event` that the subprocess sets in its `finally`
immediately after writing `return_dict`. The parent waits on this event
after `p.terminate()` instead of asserting `not p.is_alive()` within a
fixed window. Under spawn, instance/multiprocessing teardown after the
interruption can outrun a process-death check even when the
materialization data the test cares about is already ready — that timing
mismatch was the residual flake #24924 was originally chasing.

Together these turn what was a module-wide hang into a robust,
deterministic test.

## Test Plan

- `ruff format --check` / `ruff check` pass on the changed file.
- `test_ecs_pipes_interruption_forwarding` in isolation: passes.
- Full `test_ecs.py` (4 tests including the previously-wedged
`test_ecs_pipes_task_failed_to_start` and
`test_ecs_pipes_waiter_config`): **5/5 consecutive local runs pass
cleanly at ~67s each**, no 240s timeouts, no flakes.
- Watch the `dagster-aws` steps on this PR's Buildkite run.

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Internal-RevId: 8e83e4f0d0a8915c241262bcd52beac7c777376c

Author: 2 people

python_modules/libraries/dagster-aws/dagster_aws_tests/pipes_tests/test_ecs.py

@@ -136,7 +136,7 @@ def test_ecs_pipes(
         assert asset_check_executions[0].status == AssetCheckExecutionRecordStatus.SUCCEEDED
 
 
-def _materialize_asset(env, return_dict, task_started_event):
+def _materialize_asset(env, return_dict, task_started_event, materialization_done_event):
     ecs_client = boto3.client("ecs", region_name="us-east-1", endpoint_url=_MOTO_SERVER_URL)
     cloudwatch_client = boto3.client("logs", region_name="us-east-1", endpoint_url=_MOTO_SERVER_URL)
     mock_client = LocalECSMockClient(ecs_client=ecs_client, cloudwatch_client=cloudwatch_client)
@@ -173,36 +173,62 @@ def run_task_and_signal(**kwargs):
         return_dict[0] = pipes._client.describe_tasks(  # noqa
             cluster="test-cluster", tasks=[task_arn]
         )
+        # Signal that return_dict is fully populated. The parent gates on this rather
+        # than on `p.is_alive()` because spawn/instance teardown after the interruption
+        # can outrun a fixed process-death window even when the materialization data
+        # the test cares about is already ready.
+        materialization_done_event.set()
 
 
 def test_ecs_pipes_interruption_forwarding(pipes_ecs_client: PipesECSClient):
-    with multiprocessing.Manager() as manager:
+    # Use a "spawn" context so the subprocess starts a fresh interpreter and does NOT
+    # inherit this process's open file descriptors -- in particular the moto server's
+    # listening socket on localhost:5193. The child reaches moto over the network as a
+    # client, so it has no need to inherit that socket; a forked child that outlived its
+    # expected lifetime would otherwise keep the port wedged and hang every subsequent
+    # pipes test. (Linux CI defaults to fork; macOS already defaults to spawn.)
+    ctx = multiprocessing.get_context("spawn")
+    with ctx.Manager() as manager:
         return_dict = manager.dict()
-        task_started_event = multiprocessing.Event()
+        task_started_event = ctx.Event()
+        materialization_done_event = ctx.Event()
 
-        p = multiprocessing.Process(
+        p = ctx.Process(
             target=_materialize_asset,
             args=(
                 {"SLEEP_SECONDS": "10"},
                 return_dict,
                 task_started_event,
+                materialization_done_event,
             ),
         )
         p.start()
+        try:
+            if not task_started_event.wait(timeout=60):
+                raise AssertionError("Subprocess did not launch an ECS task within 60s")
 
-        if not task_started_event.wait(timeout=60):
             p.terminate()
-            p.join(timeout=30)
+
+            # Gate on the materialization-complete signal -- set by the subprocess
+            # immediately after return_dict is populated -- rather than asserting the
+            # process is fully reaped within a fixed window. Under spawn, instance and
+            # multiprocessing teardown after the interruption can outrun a process-death
+            # check even when the data we want to assert on is already available.
+            if not materialization_done_event.wait(timeout=60):
+                raise AssertionError(
+                    "Subprocess did not complete materialization within 60s of SIGTERM"
+                )
+
+            assert return_dict[0]["tasks"][0]["containers"][0]["exitCode"] == 1
+            assert return_dict[0]["tasks"][0]["stoppedReason"] == "Dagster process was interrupted"
+        finally:
+            # Defense-in-depth: spawn already prevents the child from inheriting our
+            # fds, and we waited on a real completion signal above, but never leave
+            # the subprocess alive -- a runaway child would still hold the manager
+            # connection and consume CI resources. Always reap.
             if p.is_alive():
                 p.kill()
                 p.join()
-            raise AssertionError("Subprocess did not launch an ECS task within 60s")
-
-        p.terminate()
-        p.join(timeout=30)
-        assert not p.is_alive()
-        assert return_dict[0]["tasks"][0]["containers"][0]["exitCode"] == 1
-        assert return_dict[0]["tasks"][0]["stoppedReason"] == "Dagster process was interrupted"
 
 
 class FailToStartECSMockClient(LocalECSMockClient):