[SIMPL-30451] Fan out work units so the execution target is observable

The reference graph processed every work unit inside a single op, so exactly one worker was ever reported regardless of executor. That made the guide's claim that k8s_job_executor yields several distinct contributing_hosts false, and left the reference implementations unable to demonstrate the executor choice at all. generate_work_units is now a DynamicOut and both graphs map over it, so one step is created per unit and the loosely coupled pattern dispatches one external workload per unit.

Evidence is split into contributing_workers (host and pid, differs per process) and contributing_hosts (differs only across machines), because the previous single field could not distinguish multiprocess fan-out from no fan-out. Tests now assert the mapped step keys rather than a host count, since execute_in_process ignores executor_def and cannot prove executor behaviour on its own. Adds a Windows note: multiprocess_executor did not complete during authoring and left orphaned processes.

Changelog: fixed

.gitignore

@@ -3,3 +3,5 @@
 .venv/
 .pytest_cache/
 .coverage
+
+.tmp_dagster_home_*/

README.md

@@ -53,9 +53,13 @@ uv run dagster dev -f src/distributed_execution/repository.py
 ```
 
 The Dagster UI is then available at <http://localhost:3000>. Two jobs run end to
-end on a laptop with no cluster: `tightly_coupled_local_job` and
-`loosely_coupled_subprocess_job`. The Kubernetes variants of each are documented
-in the user guide.
+end on a laptop with no cluster: `tightly_coupled_in_process_job` and
+`loosely_coupled_subprocess_job`. The Kubernetes variants of each require a
+cluster and are documented in the user guide.
+
+> `tightly_coupled_local_job` uses `multiprocess_executor`. It did not complete on
+> a Windows development machine during authoring — see the Windows note in the
+> user guide's section 5.5.
 
 ### Running tests
 

documents/user-guide/distributed-execution-guide.md

@@ -250,8 +250,20 @@ operationally:
   [yaml/tightly-coupled/rbac-step-executor.yaml](../../yaml/tightly-coupled/rbac-step-executor.yaml).
 - Per-step resource requests apply per pod, so the aggregate request for a
   fan-out step is the per-step request multiplied by concurrency.
-- `summarise_results` now reports several distinct `contributing_hosts` instead of
-  one. That output metadata is the evidence that the switch actually took effect.
+
+The evidence that the switch took effect is in `summarise_results` output
+metadata, and the two fields say different things:
+
+| Executor | `contributing_workers` | `contributing_hosts` |
+|---|---|---|
+| `in_process_executor` | 1 | 1 |
+| `multiprocess_executor` | one per unit | 1 — same machine |
+| `k8s_job_executor` | one per unit | one per unit — separate pods |
+
+This only works because `generate_work_units` is a `DynamicOut` and the graph
+does `units.map(process_work_unit).collect()`. A single op looping over all units
+internally would report one worker under *every* executor, because one step
+cannot span processes or pods.
 
 ### 5.3 Worked example — switching to loosely coupled
 
@@ -266,7 +278,7 @@ written, so the diff below is the real one.
 def distributed_execution_reference():
     target_report = report_execution_target()
     units = generate_work_units()
-    results = process_work_units(units)          # runs in-process
+    results = units.map(process_work_unit).collect()          # runs in Dagster
     return summarise_results(results, target_report)
 ```
 
@@ -277,27 +289,27 @@ def distributed_execution_reference():
 def loosely_coupled_k8s_reference():
     target_report = report_execution_target()
     units = generate_work_units()
-    results = dispatch_external_work_k8s(units)  # dispatches, then listens
+    results = units.map(dispatch_external_work_k8s).collect() # dispatches, then listens
     return summarise_results(results, target_report)
 ```
 
 The dispatching op replaces direct computation with a pipes client call:
 
 ```python
-@op(out=Out(list))
+@op(out=Out(dict))
 def dispatch_external_work_k8s(
     context: OpExecutionContext,
-    units: list,
+    unit: int,
     pipes_k8s_client: PipesK8sClient,
-) -> list:
+) -> dict:
     completed = pipes_k8s_client.run(
         context=context,
         image=PAYLOAD_IMAGE,
         command=["python", "/app/work.py"],
         namespace=PAYLOAD_NAMESPACE,
-        extras={"units": units},
+        extras={"units": [unit]},
     )
-    return _results_from_pipes(context, completed)
+    return _result_from_pipes(context, completed)
 ```
 
 and the job supplies the client as a resource instead of an executor:
@@ -343,6 +355,11 @@ else from the Dagster ecosystem — a test asserts this, because the moment the
 payload imports `dagster` the isolation argument for choosing this pattern
 collapses.
 
+Because the graph maps over the dynamic output, one external workload is
+dispatched **per unit**. A test asserts that the four units come back from four
+distinct external workers, which is the loosely coupled equivalent of the
+`contributing_workers` evidence in section 5.2.
+
 #### Message channel choice
 
 `PipesK8sClient` defaults to `PipesK8sPodLogsMessageReader`, which is what the
@@ -383,7 +400,7 @@ Those take effect only when a new image is built and the code location reloads.
 | Job | Executor / transport | Runs locally | Demonstrates |
 |---|---|---|---|
 | `tightly_coupled_in_process_job` | `in_process_executor` | Yes | Baseline; steps inside the run worker |
-| `tightly_coupled_local_job` | `multiprocess_executor` | Yes | Subprocess fan-out bounded by the run pod |
+| `tightly_coupled_local_job` | `multiprocess_executor` | Linux/macOS | Subprocess fan-out bounded by the run pod |
 | `tightly_coupled_k8s_job` | `k8s_job_executor` | No — needs a cluster | One Kubernetes Job per step |
 | `loosely_coupled_subprocess_job` | `PipesSubprocessClient` | Yes | The pipes contract end to end, no cluster |
 | `loosely_coupled_k8s_job` | `PipesK8sClient` | No — needs a cluster | External Job dispatch, messages over pod logs |
@@ -392,6 +409,14 @@ The last two dispatch the same [payload/work.py](../../payload/work.py). The
 subprocess variant exists so the pipes contract can be exercised, tested and
 demonstrated without any infrastructure.
 
+> **Windows note.** `tightly_coupled_local_job` did not complete on a Windows
+> development machine during authoring: the `multiprocess_executor` spawned step
+> subprocesses that never terminated, and required manual cleanup. This was not
+> reproduced on Linux and is not expected to affect cluster deployments, where
+> run workers are Linux pods. For a laptop demonstration on Windows, prefer
+> `tightly_coupled_in_process_job` or `loosely_coupled_subprocess_job`, both of
+> which are covered by the test suite.
+
 Run the local variants with:
 
 ```bash

documents/user-guide/readiness-checklist.md

@@ -9,8 +9,8 @@ location image.
 
 > **NOT CLUSTER-VERIFIED.** Checks L4–L9 and the Kubernetes rows of section 4.3
 > are derived from the implemented reference but have not yet been run against a
-> Simpl cluster. Checks L1–L3, which exercise the payload contract and the
-> message-parsing path, are covered by the test suite.
+> Simpl cluster. Checks L1–L3 and L10, which exercise the payload contract, the
+> message-parsing path and per-unit dispatch, are covered by the test suite.
 
 ---
 
@@ -32,9 +32,9 @@ location image.
 | T1 | Run pod reaches the metadata database | Launch `tightly_coupled_in_process_job` | Run reaches `SUCCESS`; `report_execution_target` output metadata lists `DAGSTER_POSTGRES_HOST`, `DAGSTER_POSTGRES_USER` and `DAGSTER_POSTGRES_DB` under `env_vars_present`, and `env_vars_missing` is empty |
 | T2 | Vault injection works | Same run; inspect the run pod | `kubectl -n dagster describe pod <run-pod>` shows the `vault-env` init container completed; no `vault:` literal remains in the process environment |
 | T3 | Object storage is reachable | Same run, if the workflow uses S3 | No `EndpointConnectionError` in run logs; `S3_ENDPOINT_URL` present in `env_vars_present` |
-| T4 | Multiprocess fan-out is bounded correctly | Launch `tightly_coupled_local_job` | Run succeeds; `summarise_results` metadata shows exactly one entry in `contributing_hosts`, confirming steps stayed on the run worker |
+| T4 | Multiprocess fan-out actually fans out | Launch `tightly_coupled_local_job` | Run succeeds; `summarise_results` metadata shows one entry per unit in `contributing_workers` and a single entry in `contributing_hosts` — separate processes, same machine |
 | T5 | RBAC permits step Jobs | `kubectl -n dagster auth can-i create jobs --as=system:serviceaccount:dagster:dagster-dev` | Returns `yes`; required only for `k8s_job_executor` |
-| T6 | Step pods are actually created | Launch `tightly_coupled_k8s_job`, then `kubectl -n dagster get jobs -l dagster/run-id=<run-id>` | One Job per step; `summarise_results` metadata shows **several** distinct `contributing_hosts` |
+| T6 | Step pods are actually created | Launch `tightly_coupled_k8s_job`, then `kubectl -n dagster get jobs -l dagster/run-id=<run-id>` | One Job per mapped unit; `contributing_hosts` now shows one entry **per unit**, not one |
 | T7 | Step pod egress is permitted | Same run | Steps do not hang in `STARTING`; run logs contain no connection timeouts to port 5432 |
 | T8 | Failure surfaces as a pod failure | Force a step failure in a scratch namespace | `failPodOnRunFailure: true` is set, and the step pod reports `Failed` rather than `Completed` |
 
@@ -50,10 +50,11 @@ location image.
 | L6 | Dispatcher can read pod logs — **the message channel** | `kubectl -n <payload-ns> auth can-i get pods/log --as=system:serviceaccount:dagster:dagster-dev` | Returns `yes`. A `no` here breaks reporting *without* failing the workload |
 | L7 | Payload Job is actually created | Launch `loosely_coupled_k8s_job`, then `kubectl -n <payload-ns> get jobs -l app.kubernetes.io/name=distributed-execution-payload` | One Job per dispatch, labelled `dagster/execution-target=loosely-coupled` |
 | L8 | Payload has **no** orchestration connectivity | Same run; inspect run logs | No `Payload could see orchestration runtime credentials` warning. This is a positive check — absence of errors is not sufficient |
-| L9 | Work ran off-platform | Same run | `summarise_results` metadata shows `contributing_hosts` containing the payload pod name, not the run worker's hostname |
+| L9 | Work ran off-platform | Same run | `summarise_results` metadata shows `contributing_hosts` containing the payload pod names, not the run worker's hostname |
+| L10 | One workload dispatched per unit | Same run, or `uv run pytest -k dispatched_to_its_own` locally | `contributing_workers` has one entry per unit; the local test asserts four distinct external workers |
 
-Checks L4–L9 require a cluster. L1–L3 run on a laptop and should gate every
-change to the payload or the dispatching op.
+Checks L4–L9 require a cluster. L1–L3 and L10 run on a laptop and should gate
+every change to the payload or the dispatching op.
 
 ---
 

payload/work.py

@@ -27,10 +27,13 @@ def main() -> None:
     with open_dagster_pipes() as pipes:
         units = pipes.get_extra("units")
         host = socket.gethostname()
+        worker = f"{host}#{os.getpid()}"
 
-        pipes.log.info(f"External payload started on {host} with {len(units)} work units")
+        pipes.log.info(f"External payload started on {worker} with {len(units)} work units")
 
-        results = [{"unit": unit, "squared": unit * unit, "host": host} for unit in units]
+        results = [
+            {"unit": unit, "squared": unit * unit, "host": host, "worker": worker} for unit in units
+        ]
 
         leaked = [name for name in ORCHESTRATION_ENV_VARS if os.environ.get(name)]
         if leaked:
@@ -44,11 +47,12 @@ def main() -> None:
             {
                 "results": results,
                 "host": host,
+                "worker": worker,
                 "orchestration_env_visible": leaked,
             }
         )
 
-        pipes.log.info(f"External payload finished on {host}")
+        pipes.log.info(f"External payload finished on {worker}")
 
 
 if __name__ == "__main__":

src/distributed_execution/loosely_coupled/jobs.py

@@ -14,7 +14,16 @@ import os
 import sys
 from pathlib import Path
 
-from dagster import Failure, OpExecutionContext, Out, PipesSubprocessClient, graph, op
+from dagster import (
+    Failure,
+    OpExecutionContext,
+    Out,
+    PipesSubprocessClient,
+    graph,
+    in_process_executor,
+    multiprocess_executor,
+    op,
+)
 from dagster_k8s import PipesK8sClient
 
 from distributed_execution.ops import (
@@ -39,8 +48,8 @@ COMMON_TAGS = {
 }
 
 
-def _results_from_pipes(context: OpExecutionContext, completed) -> list:
-    """Read the payload's results off the message channel.
+def _result_from_pipes(context: OpExecutionContext, completed) -> dict:
+    """Read one unit's result off the message channel.
 
     A broken message path is the defining failure mode of this pattern: the
     external workload can exit 0 while reporting nothing, so silence is treated
@@ -66,42 +75,43 @@ def _results_from_pipes(context: OpExecutionContext, completed) -> list:
             ", ".join(leaked),
         )
 
-    context.log.info("Received %s results from external host %s", len(payload["results"]), payload["host"])
-    return payload["results"]
+    row = payload["results"][0]
+    context.log.info("Unit %s computed by external worker %s", row["unit"], row["worker"])
+    return row
 
 
 @op(
-    description="Dispatches the external payload as a local subprocess and listens on the pipes channel.",
-    out=Out(list),
+    description="Dispatches one external payload as a local subprocess and listens on the pipes channel.",
+    out=Out(dict),
 )
 def dispatch_external_work_subprocess(
     context: OpExecutionContext,
-    units: list,
+    unit: int,
     pipes_subprocess_client: PipesSubprocessClient,
-) -> list:
+) -> dict:
     completed = pipes_subprocess_client.run(
         context=context,
         command=[sys.executable, PAYLOAD_SCRIPT],
-        extras={"units": units},
+        extras={"units": [unit]},
     )
-    return _results_from_pipes(context, completed)
+    return _result_from_pipes(context, completed)
 
 
 @op(
-    description="Dispatches the external payload as a Kubernetes Job and listens on the pod log stream.",
-    out=Out(list),
+    description="Dispatches one external payload as a Kubernetes Job and listens on the pod log stream.",
+    out=Out(dict),
 )
 def dispatch_external_work_k8s(
     context: OpExecutionContext,
-    units: list,
+    unit: int,
     pipes_k8s_client: PipesK8sClient,
-) -> list:
+) -> dict:
     completed = pipes_k8s_client.run(
         context=context,
         image=PAYLOAD_IMAGE,
         command=["python", "/app/work.py"],
         namespace=PAYLOAD_NAMESPACE,
-        extras={"units": units},
+        extras={"units": [unit]},
         base_pod_meta={
             "labels": {
                 "app.kubernetes.io/name": "distributed-execution-payload",
@@ -109,14 +119,14 @@ def dispatch_external_work_k8s(
             }
         },
     )
-    return _results_from_pipes(context, completed)
+    return _result_from_pipes(context, completed)
 
 
 @graph
 def loosely_coupled_subprocess_reference():
     target_report = report_execution_target()
     units = generate_work_units()
-    results = dispatch_external_work_subprocess(units)
+    results = units.map(dispatch_external_work_subprocess).collect()
     return summarise_results(results, target_report)
 
 
@@ -124,7 +134,7 @@ def loosely_coupled_subprocess_reference():
 def loosely_coupled_k8s_reference():
     target_report = report_execution_target()
     units = generate_work_units()
-    results = dispatch_external_work_k8s(units)
+    results = units.map(dispatch_external_work_k8s).collect()
     return summarise_results(results, target_report)
 
 
@@ -134,6 +144,9 @@ loosely_coupled_subprocess_job = loosely_coupled_subprocess_reference.to_job(
         "Loosely coupled via subprocess transport. Demonstrates the pipes contract end to end "
         "on a laptop, with no cluster and no metadata database access from the payload."
     ),
+    # Dispatch is I/O-bound and the real work happens in the payload, so orchestrating
+    # in one process keeps the laptop demo free of platform-specific spawn behaviour.
+    executor_def=in_process_executor,
     resource_defs={"pipes_subprocess_client": PipesSubprocessClient()},
     tags={**COMMON_TAGS, "transport": "subprocess"},
 )
@@ -144,6 +157,7 @@ loosely_coupled_k8s_job = loosely_coupled_k8s_reference.to_job(
         "Loosely coupled via Kubernetes Job dispatch. Messages return over the pod log stream. "
         "Requires a cluster and RBAC to create Jobs in the target namespace."
     ),
+    executor_def=multiprocess_executor.configured({"max_concurrent": 2}),
     resource_defs={"pipes_k8s_client": PipesK8sClient()},
     tags={**COMMON_TAGS, "transport": "k8s_pod_logs"},
 )

src/distributed_execution/ops.py

@@ -6,7 +6,7 @@ happens and *how* that fact reaches the control plane.
 
 # No `from __future__ import annotations`: it stringifies the `context` hint and
 # Dagster's op context validation then rejects it.
-from dagster import Config, MetadataValue, OpExecutionContext, Out, op
+from dagster import Config, DynamicOut, DynamicOutput, MetadataValue, OpExecutionContext, Out, op
 
 from distributed_execution.preflight import (
     TIGHTLY_COUPLED_ENV_VARS,
@@ -48,34 +48,47 @@ def report_execution_target(context: OpExecutionContext) -> dict:
     return {"identity": identity, "env_check": env_check}
 
 
-@op(description="Produces the work units that later steps fan out over.", out=Out(list))
-def generate_work_units(context: OpExecutionContext, config: WorkUnitsConfig) -> list[int]:
-    units = list(range(config.count))
-    context.log.info("Generated %s work units", len(units))
-    return units
+@op(
+    description="Fans out one dynamic output per work unit, so each unit becomes its own step.",
+    out=DynamicOut(int),
+)
+def generate_work_units(context: OpExecutionContext, config: WorkUnitsConfig):
+    context.log.info("Fanning out %s work units", config.count)
+    for unit in range(config.count):
+        yield DynamicOutput(unit, mapping_key=f"unit_{unit}")
 
 
-@op(description="Performs one unit of work per element; runs once per step-execution slot.", out=Out(list))
-def process_work_units(context: OpExecutionContext, units: list[int]) -> list[dict]:
+@op(
+    description="Performs one unit of work. One step per unit, so one process or pod per unit.",
+    out=Out(dict),
+)
+def process_work_unit(context: OpExecutionContext, unit: int) -> dict:
     identity = describe_pod_identity()
-    results = [{"unit": unit, "squared": unit * unit, "host": identity["hostname"]} for unit in units]
-    context.log.info("Processed %s units on %s", len(results), identity["hostname"])
-    return results
+    context.log.info("Processing unit %s on %s", unit, identity["worker"])
+    return {
+        "unit": unit,
+        "squared": unit * unit,
+        "host": identity["hostname"],
+        "worker": identity["worker"],
+    }
 
 
-@op(description="Aggregates results and attaches the distinct hosts that contributed.")
-def summarise_results(context: OpExecutionContext, results: list[dict], target_report: dict) -> dict:
+@op(description="Aggregates results and records which workers actually contributed.")
+def summarise_results(context: OpExecutionContext, results: list, target_report: dict) -> dict:
     hosts = sorted({row["host"] for row in results})
+    workers = sorted({row["worker"] for row in results})
     summary = {
         "units": len(results),
         "total": sum(row["squared"] for row in results),
         "contributing_hosts": hosts,
+        "contributing_workers": workers,
         "launcher_namespace": target_report["identity"]["namespace"],
     }
     context.add_output_metadata(
         {
             "units": MetadataValue.int(summary["units"]),
             "contributing_hosts": MetadataValue.json(hosts),
+            "contributing_workers": MetadataValue.json(workers),
             "launcher_namespace": MetadataValue.text(summary["launcher_namespace"]),
         }
     )

src/distributed_execution/preflight.py

@@ -52,9 +52,12 @@ def check_endpoint_reachable(url: str, timeout: float = 3.0) -> dict:
 
 def describe_pod_identity() -> dict:
     """Where this process is actually running - the primary execution-target evidence."""
+    hostname = socket.gethostname()
     return {
-        "hostname": socket.gethostname(),
+        "hostname": hostname,
         "pid": os.getpid(),
+        # Distinguishes processes on one host, so multiprocess fan-out is visible locally.
+        "worker": f"{hostname}#{os.getpid()}",
         "namespace": os.environ.get("DAGSTER_K8S_PIPELINE_RUN_NAMESPACE", "<not-in-kubernetes>"),
         "run_id": os.environ.get("DAGSTER_RUN_ID", "<unset>"),
         "image": os.environ.get("DAGSTER_K8S_PIPELINE_RUN_IMAGE", "<unset>"),

src/distributed_execution/tightly_coupled/jobs.py

@@ -17,7 +17,7 @@ from dagster_k8s import k8s_job_executor
 
 from distributed_execution.ops import (
     generate_work_units,
-    process_work_units,
+    process_work_unit,
     report_execution_target,
     summarise_results,
 )
@@ -47,7 +47,7 @@ def distributed_execution_reference():
     """Shared topology, so the three jobs differ only by execution target."""
     target_report = report_execution_target()
     units = generate_work_units()
-    results = process_work_units(units)
+    results = units.map(process_work_unit).collect()
     return summarise_results(results, target_report)
 
 

tests/test_loosely_coupled.py

@@ -55,20 +55,29 @@ def test_subprocess_job_runs_end_to_end():
     assert summary["total"] == 14
 
 
+def test_each_unit_is_dispatched_to_its_own_external_worker():
+    result = loosely_coupled_subprocess_job.execute_in_process()
+
+    mapped = result.output_for_node("dispatch_external_work_subprocess")
+    assert set(mapped) == {"unit_0", "unit_1", "unit_2", "unit_3"}
+    # Each dispatch is its own OS process, so workers differ even on a single host.
+    assert len({row["worker"] for row in mapped.values()}) == 4
+
+
 def test_payload_reports_no_orchestration_credentials(monkeypatch):
     monkeypatch.delenv("DAGSTER_POSTGRES_HOST", raising=False)
     monkeypatch.delenv("DAGSTER_POSTGRES_USER", raising=False)
     monkeypatch.delenv("DAGSTER_POSTGRES_DB", raising=False)
 
     result = loosely_coupled_subprocess_job.execute_in_process()
-    results = result.output_for_node("dispatch_external_work_subprocess")
+    mapped = result.output_for_node("dispatch_external_work_subprocess")
 
-    assert len(results) == 4
-    assert all(row["host"] for row in results)
+    assert len(mapped) == 4
+    assert all(row["worker"] for row in mapped.values())
 
 
 def test_silent_message_path_is_treated_as_failure():
-    from distributed_execution.loosely_coupled.jobs import _results_from_pipes
+    from distributed_execution.loosely_coupled.jobs import _result_from_pipes
 
     class _Silent:
         def get_custom_messages(self):
@@ -83,4 +92,4 @@ def test_silent_message_path_is_treated_as_failure():
             def info(*_args, **_kwargs): ...
 
     with pytest.raises(Failure, match="No pipes messages received"):
-        _results_from_pipes(_Ctx(), _Silent())
+        _result_from_pipes(_Ctx(), _Silent())

tests/test_tightly_coupled.py

@@ -42,8 +42,16 @@ def test_in_process_job_runs_end_to_end():
     summary = result.output_for_node("summarise_results")
     assert summary["units"] == 4
     assert summary["total"] == 14  # 0 + 1 + 4 + 9
-    # In-process execution never fans out beyond the run worker.
-    assert len(summary["contributing_hosts"]) == 1
+    # In-process execution keeps every step in the run worker's own process.
+    assert len(summary["contributing_workers"]) == 1
+
+
+def test_work_units_fan_out_into_one_step_each():
+    result = tightly_coupled_in_process_job.execute_in_process()
+
+    mapped = result.output_for_node("process_work_unit")
+    # Without this the executor choice would be meaningless: one step cannot span pods.
+    assert set(mapped) == {"unit_0", "unit_1", "unit_2", "unit_3"}
 
 
 def test_env_var_check_reports_missing_names():