fix(dagster-aws): make ECS pipes interruption test robust under CI load (#24924)

## Summary & Motivation

`test_ecs_pipes_interruption_forwarding` has been flaking on `internal`
builds when CI is under load (e.g.
https://buildkite.com/dagster/internal/builds/152968,
https://buildkite.com/dagster/internal/builds/152985), failing with
`KeyError: 0` at the final assertion.

Root cause is a race in the test itself, not a CI resource issue. The
test spawns a subprocess that runs `_materialize_asset`, then sleeps a
hardcoded 4 seconds before sending `SIGTERM` and asserting that the
subprocess populated `return_dict[0]` with the stopped ECS task state.
`return_dict[0]` is only written from a `finally` block, gated by
`assert len(pipes._client._task_runs) > 0`. Under load, Python startup +
`DagsterInstance.from_config()` + moto setup take longer than 4 seconds,
so the subprocess receives `SIGTERM` before `materialize()` has launched
the ECS task. The `finally`-block precondition fails, the child dies
without writing `return_dict[0]`, and the parent's assertion blows up.
The captured stderr in both failing builds shows the child died inside
`instance_for_test()` → `unified_storage = instance_ref.storage`, well
before any task was launched.

Replace the fixed sleep with an explicit "task launched" signal. The
child monkey-patches `mock_client.run_task` to set a
`multiprocessing.Event` as soon as `run_task` returns; the parent
`event.wait(timeout=60)` before sending `SIGTERM`, guaranteeing there is
a live ECS task to interrupt. The unbounded `while p.is_alive():
sleep(4); terminate()` retry loop is replaced with a single
`terminate()` + bounded `join(timeout=30)`.

## Test Plan

- `tox -e py312 -- -k test_ecs_pipes_interruption_forwarding` in
`python_modules/libraries/dagster-aws` (from `dagster-oss/`).
- Watch the dagster-aws step on this PR's Buildkite run; it should pass
without retries.

Internal-RevId: b8eefc49d4bf8091e176569ccbde859c273624fb

Author: smackesey

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

@@ -2,7 +2,6 @@
 import os
 import re
 import sys
-import time
 from typing import TYPE_CHECKING
 
 import boto3
@@ -137,11 +136,25 @@ def test_ecs_pipes(
         assert asset_check_executions[0].status == AssetCheckExecutionRecordStatus.SUCCEEDED
 
 
-def _materialize_asset(env, return_dict):
+def _materialize_asset(env, return_dict, task_started_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)
+
+    # Signal the parent as soon as the ECS task has actually been launched, so
+    # the parent waits for real work before sending SIGTERM rather than racing
+    # subprocess startup with a fixed sleep.
+    original_run_task = mock_client.run_task
+
+    def run_task_and_signal(**kwargs):
+        result = original_run_task(**kwargs)
+        task_started_event.set()
+        return result
+
+    mock_client.run_task = run_task_and_signal  # type: ignore[method-assign]  # ty: ignore[invalid-assignment]
+
     pipes = PipesECSClient(
-        client=LocalECSMockClient(ecs_client=ecs_client, cloudwatch_client=cloudwatch_client),  # type: ignore
+        client=mock_client,  # type: ignore
         message_reader=PipesCloudWatchMessageReader(
             client=cloudwatch_client,
         ),
@@ -165,25 +178,29 @@ def _materialize_asset(env, return_dict):
 def test_ecs_pipes_interruption_forwarding(pipes_ecs_client: PipesECSClient):
     with multiprocessing.Manager() as manager:
         return_dict = manager.dict()
+        task_started_event = multiprocessing.Event()
 
         p = multiprocessing.Process(
             target=_materialize_asset,
             args=(
                 {"SLEEP_SECONDS": "10"},
                 return_dict,
+                task_started_event,
             ),
         )
         p.start()
 
-        while p.is_alive():
-            # we started executing the run
-            # time to interrupt it!
-            time.sleep(4)
+        if not task_started_event.wait(timeout=60):
             p.terminate()
-
-        p.join()
+            p.join(timeout=30)
+            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()
-        # breakpoint()
         assert return_dict[0]["tasks"][0]["containers"][0]["exitCode"] == 1
         assert return_dict[0]["tasks"][0]["stoppedReason"] == "Dagster process was interrupted"