From 8fe51dc06f5889033336d2aa821868660e3bfbfb Mon Sep 17 00:00:00 2001 From: Joe Freeman Date: Sun, 6 Sep 2026 23:38:03 +0100 Subject: [PATCH 1/2] Expose pending dependencies from run topic --- server/lib/coflux/orchestration/server.ex | 197 +++++++++++++++++++--- server/lib/coflux/topics/run.ex | 123 +++++++++++--- tests/test_scheduling.py | 57 +++++++ tests/test_streams.py | 2 + 4 files changed, 338 insertions(+), 41 deletions(-) diff --git a/server/lib/coflux/orchestration/server.ex b/server/lib/coflux/orchestration/server.ex index c6c3bb1e..6f6836ca 100644 --- a/server/lib/coflux/orchestration/server.ex +++ b/server/lib/coflux/orchestration/server.ex @@ -1402,7 +1402,7 @@ defmodule Coflux.Orchestration.Server do child_added: child_added }} -> # Compute and register pending dependencies for non-memoised executions - {state, pending_dependencies} = + {state, pending_dependencies, argument_dependencies, unresolved_dependencies} = if step_id && !memo_hit do wait_for = Keyword.get(opts, :wait_for) || [] @@ -1412,9 +1412,11 @@ defmodule Coflux.Orchestration.Server do state = register_pending_dependencies(state, execution_id, pending_dependencies) - {state, pending_dependencies} + {state, pending_dependencies, + build_argument_dependencies(state.db, step_id, wait_for), + unresolved_dependency_ids(state.db, execution_id)} else - {state, MapSet.new()} + {state, MapSet.new(), %{}, MapSet.new()} end group_id = Keyword.get(opts, :group_id) @@ -1486,7 +1488,8 @@ defmodule Coflux.Orchestration.Server do |> notify_listeners( {:run, run.external_id}, {:execution, step_number, attempt, execution_external_id, ws_ext_id, created_at, - execute_after, %{}, nil, enrich_checkpoints(checkpoints, state.db)} + execute_after, argument_dependencies, nil, + enrich_checkpoints(checkpoints, state.db), unresolved_dependencies} ) else state @@ -2195,7 +2198,7 @@ defmodule Coflux.Orchestration.Server do state, {:run, run_external_id}, {:stream_dependency, consumer_execution_external_id, - stream_external_id(stream_run_ext_id, step_number, index), module, target} + stream_external_id(stream_run_ext_id, step_number, index), module, target, false} ) state = flush_notifications(state) @@ -2421,7 +2424,7 @@ defmodule Coflux.Orchestration.Server do state, {:run, run_external_id}, {:asset_dependency, from_execution_external_id, asset_external_id, - {asset_name, total_count, total_size, entry}} + {asset_name, total_count, total_size, entry}, false} ) else state @@ -4754,6 +4757,14 @@ defmodule Coflux.Orchestration.Server do state = register_pending_dependencies(state, execution_id, pending_dependencies) + dependencies = + Map.merge( + build_argument_dependencies(state.db, step.id, step.wait_for), + dependencies + ) + + unresolved_dependencies = unresolved_dependency_ids(state.db, execution_id) + step_requires = if step.requires_tag_set_id do {:ok, tag_set} = TagSets.get_tag_set(state.db, step.requires_tag_set_id) @@ -4794,7 +4805,8 @@ defmodule Coflux.Orchestration.Server do |> notify_listeners( {:run, run.external_id}, {:execution, step.number, attempt, execution_external_id, ws_ext_id, created_at, - execute_after, dependencies, principal, enrich_checkpoints(checkpoints, state.db)} + execute_after, dependencies, principal, enrich_checkpoints(checkpoints, state.db), + unresolved_dependencies} ) |> notify_listeners( {:modules, ws_ext_id}, @@ -4829,7 +4841,11 @@ defmodule Coflux.Orchestration.Server do state, {:run, run.external_id}, {:stream_dependency, execution_external_id, - stream_external_id(stream_run_ext_id, stream_step_number, index), module, target} + stream_external_id(stream_run_ext_id, stream_step_number, index), module, target, + MapSet.member?( + unresolved_dependencies, + stream_external_id(stream_run_ext_id, stream_step_number, index) + )} ) end) @@ -4854,7 +4870,7 @@ defmodule Coflux.Orchestration.Server do state, {:run, run.external_id}, {:input_dependency, execution_external_id, input_ext_id, input_title, - response_type} + response_type, MapSet.member?(unresolved_dependencies, input_ext_id)} ) _ -> @@ -5694,16 +5710,28 @@ defmodule Coflux.Orchestration.Server do dependencies = Map.merge( - result_deps, + build_argument_dependencies(db, step.id, step.wait_for), Map.merge( - stream_deps, + result_deps, Map.merge( - Map.get(input_deps_by_execution, execution_id, %{}), - Map.get(asset_deps_by_execution, execution_id, %{}) + stream_deps, + Map.merge( + Map.get(input_deps_by_execution, execution_id, %{}), + Map.get(asset_deps_by_execution, execution_id, %{}) + ) ) ) ) + # Nothing is outstanding for an execution that has finished: + # it isn't waiting on anything any more, whatever state its + # dependencies are in. Skipping those also keeps a large + # finished run's snapshot from re-deriving every dependency. + pending_dependencies = + if completed_at, + do: MapSet.new(), + else: unresolved_dependency_ids(db, execution_id) + {:ok, {checkpoints_before, checkpoints_after}} = Checkpoints.get_execution_snapshots( db, @@ -5727,6 +5755,7 @@ defmodule Coflux.Orchestration.Server do groups: execution_groups, assets: assets, dependencies: dependencies, + pending_dependencies: pending_dependencies, inputs: Map.get(submitted_inputs_by_execution, execution_id, %{}), result: result, result_created_by: result_created_by, @@ -7615,9 +7644,11 @@ defmodule Coflux.Orchestration.Server do |> Enum.reject(&is_nil/1) end - # Send a queue notification with the current pending dependencies for an execution. - # Converts tagged dependency keys to external IDs. - defp notify_queue_dependencies(state, execution_id, pending_dependency_ids) do + # Send a notification with the current pending dependencies for an execution. + # The queue lists the executions being waited on, as external IDs; the run + # only needs the number of gates left, so input and stream waits count + # there even though the queue drops them. + defp notify_pending_dependencies(state, execution_id, pending_dependency_ids) do case Runs.get_execution_key(state.db, execution_id) do {:ok, {r, s, a}} -> execution_ext_id = execution_external_id(r, s, a) @@ -7628,11 +7659,16 @@ defmodule Coflux.Orchestration.Server do {:ok, workspace_id} = Runs.get_workspace_id_for_execution(state.db, execution_id) ws_ext_id = workspace_external_id(state, workspace_id) - notify_listeners( - state, + state + |> notify_listeners( {:queue, ws_ext_id}, {:dependencies, execution_ext_id, dependency_ext_ids} ) + |> notify_listeners( + {:run, r}, + {:pending_dependencies, execution_ext_id, + unresolved_dependency_ids(state.db, execution_id)} + ) {:error, _} -> state @@ -7682,6 +7718,122 @@ defmodule Coflux.Orchestration.Server do end) end + # The execution references carried by the arguments named in `wait_for`, + # as {run_external_id, step_number, attempt}. These gate the execution + # before it ever runs, but live in the step's arguments rather than the + # dependency table, so nothing else surfaces them. + defp argument_reference_keys(db, step_id, wait_for) do + if wait_for && wait_for != [] do + {:ok, arguments} = Runs.get_step_arguments(db, step_id) + + wait_for + |> Enum.flat_map(fn index -> + case Enum.at(arguments, index) do + {:raw, _, references} -> references + {:blob, _, _, references} -> references + nil -> [] + end + end) + |> Enum.flat_map(fn + {:execution, run_ext, step_num, attempt} -> [{run_ext, step_num, attempt}] + _ -> [] + end) + |> Enum.uniq() + else + [] + end + end + + # Those same references, shaped as run topic dependencies. + defp build_argument_dependencies(db, step_id, wait_for) do + db + |> argument_reference_keys(step_id, wait_for) + |> Map.new(fn {run_ext, step_num, attempt} -> + ext_id = execution_external_id(run_ext, step_num, attempt) + + {module, target} = + case Runs.get_module_target(db, run_ext, step_num, attempt) do + {:ok, {m, t}} -> {m, t} + {:ok, nil} -> {nil, nil} + end + + {ext_id, {:result, {ext_id, module, target}}} + end) + end + + # Whether the execution these coordinates name has produced a result, + # following redirects (a suspend's successor, a spawn's target) the way + # the assignment gate does. + defp execution_result_pending?(db, run_ext, step_num, attempt) do + case Runs.get_execution_id(db, run_ext, step_num, attempt) do + {:ok, {execution_id}} when not is_nil(execution_id) -> + match?({:pending, _}, resolve_result(db, execution_id)) + + _ -> + false + end + end + + # Which of an execution's dependencies are still outstanding, keyed as the + # run topic's dependency map is. Related to `pending_dependencies` but not + # the same thing: that's the assignment gate, computed once and amended as + # dependencies clear, whereas this is re-derived per dependency for + # display. A completed execution reports nothing - it isn't waiting on + # anything any more, whatever state its dependencies are in. + defp unresolved_dependency_ids(db, execution_id) do + {:ok, step} = Runs.get_step_for_execution(db, execution_id) + + recorded_keys = + case Runs.get_result_dependencies(db, execution_id) do + {:ok, dependencies} -> + Enum.map(dependencies, fn {ref_id} -> + {:ok, {run_ext, step_num, attempt, _, _}} = Runs.get_execution_ref(db, ref_id) + {run_ext, step_num, attempt} + end) + end + + execution_ids = + (recorded_keys ++ argument_reference_keys(db, step.id, step.wait_for)) + |> Enum.uniq() + |> Enum.filter(fn {run_ext, step_num, attempt} -> + execution_result_pending?(db, run_ext, step_num, attempt) + end) + |> MapSet.new(fn {run_ext, step_num, attempt} -> + execution_external_id(run_ext, step_num, attempt) + end) + + input_ids = + case Runs.get_input_dependencies(db, execution_id) do + {:ok, deps} -> + deps + |> Enum.reject(fn {input_id} -> Inputs.is_input_responded?(db, input_id) end) + |> MapSet.new(fn {input_id} -> + {:ok, run_ext, number} = Inputs.get_input_run_and_number(db, input_id) + input_external_id(run_ext, number) + end) + end + + stream_ids = + case Streams.get_wait_dependencies(db, execution_id) do + {:ok, waits} -> + waits + |> Enum.filter(fn {stream_ref_id, sequence} -> + case resolve_stream_ref_id(db, stream_ref_id) do + {:ok, stream_id} -> !stream_reached?(db, stream_id, sequence) + {:error, :not_found} -> false + end + end) + |> MapSet.new(fn {stream_ref_id, _sequence} -> + {:ok, {run_ext, step_number, index, _module, _target}} = + Streams.get_stream_ref(db, stream_ref_id) + + stream_external_id(run_ext, step_number, index) + end) + end + + execution_ids |> MapSet.union(input_ids) |> MapSet.union(stream_ids) + end + # Compute the set of execution IDs that the given execution is waiting on. # This covers both argument references (when wait_for is set) and result_dependencies. defp compute_pending_dependencies(db, execution_id, wait_for, step_id) do @@ -8038,7 +8190,7 @@ defmodule Coflux.Orchestration.Server do put_in(state, [Access.key(:pending_dependencies), waiter_id], updated) end end) - |> notify_queue_dependencies(waiter_id, updated) + |> notify_pending_dependencies(waiter_id, updated) :error -> state @@ -8112,7 +8264,7 @@ defmodule Coflux.Orchestration.Server do put_in(state, [Access.key(:pending_dependencies), waiter_id], updated) end end) - |> notify_queue_dependencies(waiter_id, updated) + |> notify_pending_dependencies(waiter_id, updated) :error -> state @@ -8264,7 +8416,8 @@ defmodule Coflux.Orchestration.Server do notify_listeners( state, {:run, run_external_id}, - {:result_dependency, from_execution_external_id, dep_ext_id, dependency} + {:result_dependency, from_execution_external_id, dep_ext_id, dependency, + match?({:pending, _}, resolve_result(state.db, execution_id))} ) else state @@ -8364,7 +8517,7 @@ defmodule Coflux.Orchestration.Server do |> notify_listeners( {:run, run_external_id}, {:input_dependency, from_execution_external_id, input_external_id, title, - response_type} + response_type, is_nil(response_type)} ) |> notify_listeners( {:inputs, ws_ext_id}, diff --git a/server/lib/coflux/topics/run.ex b/server/lib/coflux/topics/run.ex index 5b7418d8..ad639c26 100644 --- a/server/lib/coflux/topics/run.ex +++ b/server/lib/coflux/topics/run.ex @@ -76,7 +76,7 @@ defmodule Coflux.Topics.Run do defp process_notification( topic, {:execution, step_number, attempt, execution_external_id, workspace_external_id, - created_at, execute_after, dependencies, created_by, checkpoints} + created_at, execute_after, dependencies, created_by, checkpoints, pending_dependencies} ) do if workspace_external_id in topic.state.workspace_ids do Topic.set( @@ -100,7 +100,11 @@ defmodule Coflux.Topics.Run do assets: %{}, dependencies: Map.new(dependencies, fn {dependency_id, dependency} -> - {dependency_id, build_dependency(dependency)} + {dependency_id, + build_dependency( + dependency, + MapSet.member?(pending_dependencies, dependency_id) + )} end), children: [], inputs: %{}, @@ -162,11 +166,25 @@ defmodule Coflux.Topics.Run do end) end + # Which dependencies the execution is still waiting on. Sent as the whole + # set rather than a delta: a result that redirects (to a suspend's + # successor, say) clears a dependency keyed by the execution originally + # referenced, so there's no dependable one-to-one between what resolved + # and which entry it releases. defp process_notification( topic, - {:result_dependency, execution_external_id, dependency_id, dependency} + {:pending_dependencies, execution_external_id, pending} ) do - dependency = build_dependency(dependency) + set_pending_dependencies(topic, execution_external_id, fn dependency_id -> + MapSet.member?(pending, dependency_id) + end) + end + + defp process_notification( + topic, + {:result_dependency, execution_external_id, dependency_id, dependency, pending} + ) do + dependency = build_dependency(dependency, pending) update_execution( topic, @@ -183,9 +201,15 @@ defmodule Coflux.Topics.Run do defp process_notification( topic, - {:stream_dependency, execution_external_id, stream_id, module, target} + {:stream_dependency, execution_external_id, stream_id, module, target, pending} ) do - dependency = %{type: "stream", streamId: stream_id, module: module, target: target} + dependency = %{ + type: "stream", + streamId: stream_id, + module: module, + target: target, + pending: pending + } update_execution(topic, execution_external_id, fn topic, base_path -> Topic.merge(topic, base_path ++ [:dependencies, stream_id], dependency) @@ -217,7 +241,8 @@ defmodule Coflux.Topics.Run do topic, {:completion, execution_external_id, kind, successor, completion_at} ) do - update_execution(topic, execution_external_id, fn topic, base_path -> + topic + |> update_execution(execution_external_id, fn topic, base_path -> topic |> Topic.set(base_path ++ [:completedAt], completion_at) |> Topic.set(base_path ++ [:completion], %{ @@ -225,6 +250,9 @@ defmodule Coflux.Topics.Run do successor: successor }) end) + # Nothing is outstanding once the execution has finished, whatever state + # its dependencies are in - the rule the snapshot applies too. + |> set_pending_dependencies(execution_external_id, fn _ -> false end) end defp process_notification(topic, {:checkpoints, execution_external_id, checkpoints}) do @@ -328,27 +356,30 @@ defmodule Coflux.Topics.Run do defp process_notification( topic, - {:input_dependency, execution_external_id, input_external_id, title, response_type} + {:input_dependency, execution_external_id, input_external_id, title, response_type, + pending} ) do update_execution(topic, execution_external_id, fn topic, base_path -> Topic.set(topic, base_path ++ [:dependencies, input_external_id], %{ type: "input", inputId: input_external_id, title: title, - status: response_type + status: response_type, + pending: pending }) end) end defp process_notification( topic, - {:asset_dependency, execution_external_id, asset_external_id, asset} + {:asset_dependency, execution_external_id, asset_external_id, asset, pending} ) do update_execution(topic, execution_external_id, fn topic, base_path -> Topic.set(topic, base_path ++ [:dependencies, asset_external_id], %{ type: "asset", assetId: asset_external_id, - asset: build_asset(asset) + asset: build_asset(asset), + pending: pending }) end) end @@ -388,6 +419,27 @@ defmodule Coflux.Topics.Run do end) end + defp set_pending_dependencies(topic, execution_external_id, pending?) do + Enum.reduce(topic.value.steps, topic, fn {step_id, step}, topic -> + Enum.reduce(step.executions, topic, fn {attempt, execution}, topic -> + if execution.executionId == execution_external_id do + execution + |> Map.get(:dependencies, %{}) + |> Map.keys() + |> Enum.reduce(topic, fn dependency_id, topic -> + Topic.set( + topic, + [:steps, step_id, :executions, attempt, :dependencies, dependency_id, :pending], + pending?.(dependency_id) + ) + end) + else + topic + end + end) + end) + end + defp update_dependency_input_status(topic, input_external_id, response_type) do Enum.reduce(topic.value.steps, topic, fn {step_id, step}, topic -> Enum.reduce(step.executions, topic, fn {attempt, execution}, topic -> @@ -459,7 +511,11 @@ defmodule Coflux.Topics.Run do Map.new(execution.assets, fn {external_asset_id, asset} -> {external_asset_id, build_asset(asset)} end), - dependencies: build_dependencies(execution.dependencies), + dependencies: + build_dependencies( + execution.dependencies, + Map.get(execution, :pending_dependencies, MapSet.new()) + ), children: Enum.map(execution.children, &build_child(&1, run.external_id)), inputs: Map.get(execution, :inputs, %{}), result: build_result(execution.result, execution.result_created_by), @@ -490,26 +546,55 @@ defmodule Coflux.Topics.Run do } end - defp build_dependencies(dependencies) do + # `pending` marks a dependency the execution is still waiting on. It's + # carried per entry rather than as a count so the graph can point at the + # one that's holding a step back. + defp build_dependencies(dependencies, pending) do Map.new(dependencies, fn {id, {:result, execution}} -> - {id, %{type: "result", execution: build_execution(execution)}} + {id, + %{ + type: "result", + execution: build_execution(execution), + pending: MapSet.member?(pending, id) + }} {id, {:input, title, status}} -> - {id, %{type: "input", inputId: id, title: title, status: status}} + {id, + %{ + type: "input", + inputId: id, + title: title, + status: status, + pending: MapSet.member?(pending, id) + }} {id, {:asset, asset}} -> - {id, %{type: "asset", assetId: id, asset: build_asset(asset)}} + {id, + %{ + type: "asset", + assetId: id, + asset: build_asset(asset), + pending: MapSet.member?(pending, id) + }} {id, {:stream, stream_id, module, target}} -> - {id, %{type: "stream", streamId: stream_id, module: module, target: target}} + {id, + %{ + type: "stream", + streamId: stream_id, + module: module, + target: target, + pending: MapSet.member?(pending, id) + }} end) end - defp build_dependency(execution) do + defp build_dependency(execution, pending) do %{ type: "result", - execution: build_execution(execution) + execution: build_execution(execution), + pending: pending } end diff --git a/tests/test_scheduling.py b/tests/test_scheduling.py index e64cda30..436b3b1b 100644 --- a/tests/test_scheduling.py +++ b/tests/test_scheduling.py @@ -109,6 +109,63 @@ def test_wait_for(worker): assert ctx.result(run_id)["value"]["data"] == "all done" +def test_pending_dependencies_exposed_in_run_topic(worker): + """The run topic marks the dependencies an execution is still waiting on.""" + targets = [ + workflow("test", "main"), + task("test", "producer"), + task("test", "consumer", parameters=["data"], wait_for=[0]), + ] + + with worker(targets, concurrency=3) as ctx: + resp = ctx.submit("test", "main") + run_id = resp["runId"] + + ex0 = ctx.executor.next_execute() + ref_a = ex0.conn.submit_task(ex0.execution_id, "test", "producer", []) + consumer_args = [ + { + "type": "inline", + "format": "json", + "value": None, + "references": [["execution", ref_a]], + } + ] + ex0.conn.submit_task( + ex0.execution_id, "test", "consumer", consumer_args, wait_for=[0] + ) + + ex1 = ctx.executor.next_execute() + assert ex1.target == "producer" + + # The consumer is held behind the producer. That gate comes from a + # wait_for argument reference rather than a recorded dependency, but + # it still shows up among the dependencies, marked pending. + steps = ctx.inspect(run_id)["steps"] + producer_id = next(s for s in steps.values() if s["target"] == "producer")[ + "executions" + ]["1"]["executionId"] + consumer = next(s for s in steps.values() if s["target"] == "consumer") + dependency = consumer["executions"]["1"]["dependencies"][producer_id] + assert dependency["type"] == "result" + assert dependency["pending"] is True + + ex1.conn.complete(ex1.execution_id, value=42) + + ex2 = ctx.executor.next_execute(timeout=5) + assert ex2.target == "consumer" + + # Released once the producer has a result. + steps = ctx.inspect(run_id)["steps"] + consumer = next(s for s in steps.values() if s["target"] == "consumer") + released = consumer["executions"]["1"]["dependencies"][producer_id] + assert released["pending"] is False + + ex2.conn.complete(ex2.execution_id, value="done") + ex0.conn.complete(ex0.execution_id, value="all done") + ctx.result(run_id) + + def test_wait_for_multiple_dependencies(worker): """Task with wait_for on two args waits for both referenced executions.""" targets = [ diff --git a/tests/test_streams.py b/tests/test_streams.py index 61de02d5..e29b1a8a 100644 --- a/tests/test_streams.py +++ b/tests/test_streams.py @@ -2178,12 +2178,14 @@ def test_stream_dependency_reported_for_each_consumer_attempt(worker): prod_ex.conn.stream_close(prod_ex.execution_id, 0) prod_ex.conn.complete(prod_ex.execution_id, value="done") + # Both attempts have finished by now, so neither is still waiting. expected = { stream["id"]: { "type": "stream", "streamId": stream["id"], "module": "test", "target": "producer", + "pending": False, } } _, step = next(iter(ctx.inspect(cons_resp["runId"])["steps"].items())) From e22aebe5e3c1d7bc9147505edb9dec1e8fe36ea9 Mon Sep 17 00:00:00 2001 From: Joe Freeman Date: Mon, 7 Sep 2026 00:12:18 +0100 Subject: [PATCH 2/2] Fix test --- tests/test_streams.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/test_streams.py b/tests/test_streams.py index e29b1a8a..5586aa23 100644 --- a/tests/test_streams.py +++ b/tests/test_streams.py @@ -545,6 +545,8 @@ def test_unsubscribe_prevents_receiving_full_stream(worker): consumer if they're in flight when the server processes unsubscribe. The meaningful check is that the consumer stops seeing items before the full stream is delivered — not that unsubscribe is synchronous. + The barrier below narrows that window rather than closing it, so the + tolerance stays. """ targets = [workflow("test", "producer"), workflow("test", "consumer")] @@ -566,6 +568,12 @@ def test_unsubscribe_prevents_receiving_full_stream(worker): first = cons_ex.conn.recv_push("stream_items", subscription_id=1, timeout=3) assert first["items"][0][1]["value"] == 0 cons_ex.conn.stream_unsubscribe(cons_ex.execution_id, subscription_id=1) + # Unsubscribe is a notification and the producer has its own + # connection, so without a barrier the appends below can be handled + # first and the consumer sees the whole stream. Flush is a request: + # returning from it means the unsubscribe ahead of it on this + # connection has been processed and forwarded. + cons_ex.conn.flush(cons_ex.execution_id) # Producer keeps appending after unsubscribe. for i in range(1, 10):