diff --git a/cli/cmd/coflux/logs.go b/cli/cmd/coflux/logs.go index 4bf6c798..2803131a 100644 --- a/cli/cmd/coflux/logs.go +++ b/cli/cmd/coflux/logs.go @@ -106,6 +106,14 @@ func resolveStepAttempt(data map[string]any, stepAttempt string) (string, error) if !ok { return "", fmt.Errorf("step %q has invalid data", stepID) } + // Every attempt of a step is summarised under "attempts"; only the + // expanded one is under "executions". + attempts, _ := s["attempts"].(map[string]any) + if a, ok := attempts[attempt].(map[string]any); ok { + if eid, ok := a["executionId"].(string); ok { + return eid, nil + } + } executions, _ := s["executions"].(map[string]any) execData, ok := executions[attempt] if !ok { diff --git a/server/lib/coflux/application.ex b/server/lib/coflux/application.ex index fda04af5..f8601034 100644 --- a/server/lib/coflux/application.ex +++ b/server/lib/coflux/application.ex @@ -37,6 +37,9 @@ defmodule Coflux.Application do Topics.Workspaces, Topics.Modules, Topics.Run, + Topics.RunExecution, + Topics.RunGroup, + Topics.RunSteps, Topics.Workflow, Topics.Pools, Topics.Pool, diff --git a/server/lib/coflux/orchestration.ex b/server/lib/coflux/orchestration.ex index 390ff2f1..3e961ff2 100644 --- a/server/lib/coflux/orchestration.ex +++ b/server/lib/coflux/orchestration.ex @@ -316,6 +316,13 @@ defmodule Coflux.Orchestration do call_server(project_id, {:subscribe_run, run_id, pid}) end + # Per-execution detail (results, dependencies, checkpoints, assets...) and + # per-step arguments and streams, for the parts of a run a topic shows. + # `request` is `%{executions: [{step_number, attempt}], steps: [step_number]}`. + def get_run_details(project_id, run_id, request) do + call_server(project_id, {:get_run_details, run_id, request}) + end + def subscribe_stream_topic(project_id, stream_id, pid) do call_server(project_id, {:subscribe_stream_topic, stream_id, pid}) end diff --git a/server/lib/coflux/orchestration/results.ex b/server/lib/coflux/orchestration/results.ex index 06b98a8d..16117bda 100644 --- a/server/lib/coflux/orchestration/results.ex +++ b/server/lib/coflux/orchestration/results.ex @@ -372,6 +372,44 @@ defmodule Coflux.Orchestration.Results do end end + # Every completion in the run, with the successor resolved to the + # coordinates of the execution it names (an execution in this database, + # or an execution ref). Shape: + # `{:ok, %{execution_id => {kind_atom, {run_ext, step_number, attempt} | nil, created_at}}}` + def get_run_completions(db, run_id) do + case query( + db, + """ + SELECT c.execution_id, c.kind, c.created_at, + r2.external_id, s2.number, e2.attempt, + ref.run_external_id, ref.step_number, ref.attempt + FROM completions AS c + INNER JOIN executions AS e ON e.id = c.execution_id + INNER JOIN steps AS s ON s.id = e.step_id + LEFT JOIN executions AS e2 ON e2.id = c.successor_id + LEFT JOIN steps AS s2 ON s2.id = e2.step_id + LEFT JOIN runs AS r2 ON r2.id = s2.run_id + LEFT JOIN execution_refs AS ref ON ref.id = c.successor_ref_id + WHERE s.run_id = ?1 + """, + {run_id} + ) do + {:ok, rows} -> + {:ok, + Map.new(rows, fn {execution_id, kind, created_at, run_ext, step_number, attempt, + ref_run_ext, ref_step_number, ref_attempt} -> + successor = + cond do + run_ext -> {run_ext, step_number, attempt} + ref_run_ext -> {ref_run_ext, ref_step_number, ref_attempt} + true -> nil + end + + {execution_id, {kind_atom(kind), successor, created_at}} + end)} + end + end + # Returns the raw completion row for an execution. Shape: # `{:ok, {kind_atom, successor_id, successor_ref_id, created_at, created_by}}` # `{:ok, nil}` — no completion yet diff --git a/server/lib/coflux/orchestration/server.ex b/server/lib/coflux/orchestration/server.ex index 1468c883..57409e70 100644 --- a/server/lib/coflux/orchestration/server.ex +++ b/server/lib/coflux/orchestration/server.ex @@ -2997,8 +2997,8 @@ defmodule Coflux.Orchestration.Server do end def handle_call({:subscribe_run, external_run_id, pid}, _from, state) do - case find_and_build_run_data(state, external_run_id) do - {:ok, run, parent, steps} -> + case find_run(state, external_run_id, &build_run_structure/2) do + {:ok, {run, parent, steps}} -> {:ok, ref, state} = add_listener(state, {:run, run.external_id}, pid) {:reply, {:ok, run, parent, steps, ref}, state} @@ -3007,6 +3007,13 @@ defmodule Coflux.Orchestration.Server do end end + def handle_call({:get_run_details, external_run_id, request}, _from, state) do + case find_run(state, external_run_id, &build_run_details(&1, &2, request)) do + {:ok, details} -> {:reply, {:ok, details}, state} + :not_found -> {:reply, {:error, :not_found}, state} + end + end + def handle_call({:subscribe_stream_topic, stream_external_id, pid}, _from, state) do case build_stream_topic_initial(state, stream_external_id) do {:ok, initial} -> @@ -5621,16 +5628,18 @@ defmodule Coflux.Orchestration.Server do end end - defp find_and_build_run_data(state, external_run_id) do + # Runs `fun.(db, run)` against the database the run lives in: the active + # epoch, or an archived one found through the epoch index. + defp find_run(state, external_run_id, fun) do case Runs.get_run_by_external_id(state.db, external_run_id) do {:ok, run} when not is_nil(run) -> - build_run_data(state.db, run) + {:ok, fun.(state.db, run)} {:ok, nil} -> query_fn = fn archive_db -> case Runs.get_run_by_external_id(archive_db, external_run_id) do {:ok, run} when not is_nil(run) -> - {:found, build_run_data(archive_db, run)} + {:found, fun.(archive_db, run)} {:ok, nil} -> :not_found @@ -5642,182 +5651,96 @@ defmodule Coflux.Orchestration.Server do end case search_archived_epochs(state, query_fn, bloom_fn) do - {:found, result} -> result + {:found, result} -> {:ok, result} :not_found -> :not_found end end end - defp build_run_data(db, run) do + # The run's structure: every step and attempt with its status, the links + # between them, and groups — but none of the per-execution detail + # (results, dependencies, checkpoints, assets, inputs, metrics) or the + # per-step arguments and streams. Those are loaded for the parts a topic + # shows, through `build_run_details/3`. Nothing here is resolved per + # execution: external ids come from the run, step number and attempt. + defp build_run_structure(db, run) do parent = if run.parent_ref_id do resolve_execution_ref(db, run.parent_ref_id) end - run_requires = - if run.requires_tag_set_id do - case TagSets.get_tag_set(db, run.requires_tag_set_id) do - {:ok, tag_set} -> tag_set - end - else - %{} - end - - run = Map.put(run, :requires, run_requires) + run = Map.put(run, :requires, get_tag_set(db, run.requires_tag_set_id)) {:ok, steps} = Runs.get_run_steps(db, run.id) {:ok, run_executions} = Runs.get_run_executions(db, run.id) - {:ok, run_dependencies} = Runs.get_run_dependencies(db, run.id) - {:ok, run_stream_dependencies} = Streams.get_run_dependencies(db, run.id) - {:ok, run_streams} = Streams.get_streams_for_run(db, run.id) - streams_by_step = build_run_streams(db, run_streams) {:ok, run_children} = Runs.get_run_children(db, run.id) {:ok, groups} = Runs.get_groups_for_run(db, run.id) - {:ok, run_metric_defs} = Runs.get_run_metric_definitions(db, run.id) - {:ok, run_input_deps} = Inputs.get_input_dependencies_for_run(db, run.id) - {:ok, run_submitted_inputs} = Inputs.get_submitted_inputs_for_run(db, run.id) - {:ok, run_asset_deps} = Runs.get_asset_dependencies_for_run(db, run.id) - - # Resolving a checkpoint needs the workspace chain of the execution - # reading it. Resolved from `db` rather than `state` because this also - # runs against archived epochs, which remap workspace ids. - workspace_chains = - run_executions - |> Enum.map(fn {_execution_id, _step_id, _attempt, workspace_id, _, _, _, _, _} -> - workspace_id - end) - |> Enum.uniq() - |> Map.new(fn workspace_id -> - {:ok, chain} = Workspaces.get_workspace_chain(db, workspace_id) - {workspace_id, chain} - end) + {:ok, completions} = Results.get_run_completions(db, run.id) + {:ok, workspaces} = Workspaces.get_all_workspaces(db) - submitted_inputs_by_execution = - run_submitted_inputs - |> Enum.group_by( - fn {execution_id, _run_ext_id, _input_number, _title, _response_type} -> execution_id end, - fn {_execution_id, run_ext_id, input_number, title, response_type} -> - {input_external_id(run_ext_id, input_number), - %{ - title: title, - status: if(response_type, do: decode_input_response_type(response_type)) - }} - end - ) - |> Map.new(fn {execution_id, inputs} -> {execution_id, Map.new(inputs)} end) - - input_deps_by_execution = - run_input_deps - |> Enum.group_by( - fn {execution_id, _run_ext_id, _input_number, _key, _prompt_id, _title, _created_at, - _response_type, _response_value, _responded_at, _created_by} -> - execution_id - end, - fn {_execution_id, run_ext_id, input_number, _key, _prompt_id, title, _created_at, - response_type, _response_value, _responded_at, _response_created_by} -> - {input_external_id(run_ext_id, input_number), - {:input, title, if(response_type, do: decode_input_response_type(response_type))}} - end - ) - |> Map.new(fn {execution_id, deps} -> {execution_id, Map.new(deps)} end) - - asset_deps_by_execution = - run_asset_deps - |> Enum.group_by( - fn {execution_id, _asset_id} -> execution_id end, - fn {_execution_id, asset_id} -> - {external_id, name, total_count, total_size, entry} = resolve_asset(db, asset_id) - {external_id, {:asset, {name, total_count, total_size, entry}}} - end - ) - |> Map.new(fn {execution_id, deps} -> {execution_id, Map.new(deps)} end) - - metric_definitions_by_execution = - Enum.group_by( - run_metric_defs, - fn {execution_id, _, _, _, _, _, _, _, _, _, _} -> execution_id end, - fn {_, key, group, group_units, group_lower, group_upper, scale, units, progress, lower, - upper} -> - {key, - %{ - group: group, - group_units: group_units, - group_lower: group_lower, - group_upper: group_upper, - scale: scale, - units: units, - progress: progress == 1, - lower: lower, - upper: upper - }} - end - ) - |> Map.new(fn {execution_id, defs} -> {execution_id, Map.new(defs)} end) + steps_by_id = Map.new(steps, &{&1.id, &1}) - cache_configs = - steps - |> Enum.map(& &1.cache_config_id) - |> Enum.reject(&is_nil/1) - |> Enum.uniq() - |> Enum.reduce(%{}, fn cache_config_id, cache_configs -> - case CacheConfigs.get_cache_config(db, cache_config_id) do - {:ok, cache_config} -> Map.put(cache_configs, cache_config_id, cache_config) - end + external_ids = + Map.new(run_executions, fn {execution_id, step_id, attempt, _, _, _, _, _, _} -> + {execution_id, + execution_external_id(run.external_id, Map.fetch!(steps_by_id, step_id).number, attempt)} end) - results = - run_executions - |> Enum.map(&elem(&1, 0)) - |> Enum.reduce(%{}, fn execution_id, results -> - {result, result_at, completed_at, result_created_by} = - case Results.get_result(db, execution_id) do - {:ok, {result, result_at, completion_at, created_by}} -> - result = build_result(result, db) - {result, result_at, completion_at, created_by} - - {:ok, nil} -> - {nil, nil, nil, nil} - end - - completion = - case Results.get_completion(db, execution_id) do - {:ok, {kind, successor_id, successor_ref_id, _, _}} -> - %{ - kind: Atom.to_string(kind), - successor: build_completion_successor(db, successor_id, successor_ref_id) - } - - {:ok, nil} -> - nil - end - - Map.put( - results, - execution_id, - {result, result_at, completed_at, result_created_by, completion} - ) + groups_by_execution = + groups + |> Enum.group_by(&elem(&1, 0)) + |> Map.new(fn {execution_id, rows} -> + {execution_id, + Map.new(rows, fn {_, group_id, name, concurrency} -> + {group_id, %{name: name, concurrency: concurrency}} + end)} end) + executions_by_step = Enum.group_by(run_executions, &elem(&1, 1)) + cache_configs = load_cache_configs(db, steps) + steps = Map.new(steps, fn step -> - {:ok, arguments} = Runs.get_step_arguments(db, step.id) - arguments = Enum.map(arguments, &build_value(&1, db)) - - requires = - if step.requires_tag_set_id do - case TagSets.get_tag_set(db, step.requires_tag_set_id) do - {:ok, requires} -> requires - end - else - %{} - end - + # A step's parent is an execution of this run, except for a step + # copied in from an archive, which is looked up. parent_execution_external_id = if step.parent_id do - {:ok, {r, s, a}} = Runs.get_execution_key(db, step.parent_id) - execution_external_id(r, s, a) + Map.get_lazy(external_ids, step.parent_id, fn -> + {:ok, {r, s, a}} = Runs.get_execution_key(db, step.parent_id) + execution_external_id(r, s, a) + end) end + executions = + executions_by_step + |> Map.get(step.id, []) + |> Map.new(fn {execution_id, _step_id, attempt, workspace_id, execute_after, created_at, + assigned_at, created_by_user_ext_id, created_by_token_ext_id} -> + {completed_at, completion} = + case Map.get(completions, execution_id) do + nil -> + {nil, nil} + + {kind, successor, completion_at} -> + {completion_at, + %{kind: Atom.to_string(kind), successor: build_successor(successor)}} + end + + {attempt, + %{ + execution_id: Map.fetch!(external_ids, execution_id), + workspace_id: Map.fetch!(workspaces, workspace_id).external_id, + created_at: created_at, + created_by: build_created_by(created_by_user_ext_id, created_by_token_ext_id), + execute_after: execute_after, + assigned_at: assigned_at, + completed_at: completed_at, + completion: completion, + groups: Map.get(groups_by_execution, execution_id, %{}), + children: Map.get(run_children, execution_id, []) + }} + end) + {step.number, %{ module: step.module, @@ -5838,136 +5761,283 @@ defmodule Coflux.Orchestration.Server do recurrent: step.recurrent, timeout: step.timeout, created_at: step.created_at, - arguments: arguments, - requires: requires, - streams: Map.get(streams_by_step, step.id, %{}), - executions: - run_executions - |> Enum.filter(&(elem(&1, 1) == step.id)) - |> Map.new(fn {execution_id, _step_id, attempt, workspace_id, execute_after, - created_at, assigned_at, created_by_user_ext_id, - created_by_token_ext_id} -> - execution_created_by = - case {created_by_user_ext_id, created_by_token_ext_id} do - {nil, nil} -> nil - {user_ext_id, nil} -> %{type: "user", external_id: user_ext_id} - {nil, token_ext_id} -> %{type: "token", external_id: token_ext_id} - end - - {:ok, {r, s, a}} = Runs.get_execution_key(db, execution_id) - exec_external_id = execution_external_id(r, s, a) - - {:ok, workspace_external_id} = - Workspaces.get_workspace_external_id(db, workspace_id) - - {result, result_at, completed_at, result_created_by, completion} = - Map.fetch!(results, execution_id) - - execution_groups = - groups - |> Enum.filter(fn {e_id, _, _, _} -> e_id == execution_id end) - |> Map.new(fn {_, group_id, name, concurrency} -> - {group_id, %{name: name, concurrency: concurrency}} - end) - - # TODO: load assets in one query - {:ok, asset_ids} = Results.get_assets_for_execution(db, execution_id) - - assets = - asset_ids - |> Enum.map(&resolve_asset(db, &1)) - |> Map.new(fn {external_id, name, total_count, total_size, entry} -> - {external_id, {name, total_count, total_size, entry}} - end) - - # TODO: batch? get `get_dependencies` to resolve? - result_deps = - run_dependencies - |> Map.get(execution_id, []) - |> Map.new(fn dependency_ref_id -> - {ext_id, _module, _target} = - execution = - resolve_execution_ref(db, dependency_ref_id) - - {ext_id, {:result, execution}} - end) - - stream_deps = - run_stream_dependencies - |> Map.get(execution_id, []) - |> Map.new(fn stream_ref_id -> - {:ok, {stream_run_ext_id, step_number, index, module, target}} = - Streams.get_stream_ref(db, stream_ref_id) - - id = stream_external_id(stream_run_ext_id, step_number, index) - {id, {:stream, id, module, target}} - end) - - dependencies = - Map.merge( - build_argument_dependencies(db, step.id, step.wait_for), - Map.merge( - result_deps, - Map.merge( - 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, - execution_id, - step.id, - Map.fetch!(workspace_chains, workspace_id), - attempt - ) - - {attempt, - %{ - execution_id: exec_external_id, - workspace_id: workspace_external_id, - created_at: created_at, - created_by: execution_created_by, - execute_after: execute_after, - assigned_at: assigned_at, - result_at: result_at, - completed_at: completed_at, - completion: completion, - 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, - children: Map.get(run_children, execution_id, []), - metric_definitions: Map.get(metric_definitions_by_execution, execution_id, %{}), - checkpoints: %{ - before: enrich_checkpoints(checkpoints_before, db), - after: enrich_checkpoints(checkpoints_after, db) - } - }} - end) + requires: get_tag_set(db, step.requires_tag_set_id), + executions: executions }} end) - {:ok, run, parent, steps} + {run, parent, steps} end + # The detail of the executions named by `{step number, attempt}`, and the + # arguments and streams of the named steps. The row scans are over the + # whole run (they're indexed by it and cheap); everything that resolves a + # value, ref or asset is done only for what was asked for. + defp build_run_details(db, run, %{executions: execution_keys, steps: step_numbers}) do + {:ok, steps} = Runs.get_run_steps(db, run.id) + {:ok, run_executions} = Runs.get_run_executions(db, run.id) + steps_by_id = Map.new(steps, &{&1.id, &1}) + steps_by_number = Map.new(steps, &{&1.number, &1}) + + executions_by_key = + Map.new(run_executions, fn {execution_id, step_id, attempt, workspace_id, _, _, _, _, _} -> + {{Map.fetch!(steps_by_id, step_id).number, attempt}, {execution_id, workspace_id}} + end) + + requested = + Enum.flat_map(execution_keys, fn {number, attempt} -> + case Map.fetch(executions_by_key, {number, attempt}) do + {:ok, {execution_id, workspace_id}} -> + [{execution_id, workspace_id, Map.fetch!(steps_by_number, number), attempt}] + + :error -> + [] + end + end) + + requested_ids = MapSet.new(requested, &elem(&1, 0)) + requested? = fn execution_id -> MapSet.member?(requested_ids, execution_id) end + + executions = + if requested == [] do + %{} + else + {:ok, run_dependencies} = Runs.get_run_dependencies(db, run.id) + {:ok, run_stream_dependencies} = Streams.get_run_dependencies(db, run.id) + {:ok, run_metric_defs} = Runs.get_run_metric_definitions(db, run.id) + {:ok, run_input_deps} = Inputs.get_input_dependencies_for_run(db, run.id) + {:ok, run_submitted_inputs} = Inputs.get_submitted_inputs_for_run(db, run.id) + {:ok, run_asset_deps} = Runs.get_asset_dependencies_for_run(db, run.id) + + # Resolving a checkpoint needs the workspace chain of the execution + # reading it. Resolved from `db` rather than `state` because this + # also runs against archived epochs, which remap workspace ids. + workspace_chains = + requested + |> Enum.map(&elem(&1, 1)) + |> Enum.uniq() + |> Map.new(fn workspace_id -> + {:ok, chain} = Workspaces.get_workspace_chain(db, workspace_id) + {workspace_id, chain} + end) + + submitted_inputs_by_execution = + run_submitted_inputs + |> Enum.filter(fn {execution_id, _, _, _, _} -> requested?.(execution_id) end) + |> Enum.group_by( + fn {execution_id, _run_ext_id, _input_number, _title, _response_type} -> + execution_id + end, + fn {_execution_id, run_ext_id, input_number, title, response_type} -> + {input_external_id(run_ext_id, input_number), + %{ + title: title, + status: if(response_type, do: decode_input_response_type(response_type)) + }} + end + ) + |> Map.new(fn {execution_id, inputs} -> {execution_id, Map.new(inputs)} end) + + input_deps_by_execution = + run_input_deps + |> Enum.filter(fn row -> requested?.(elem(row, 0)) end) + |> Enum.group_by( + fn {execution_id, _run_ext_id, _input_number, _key, _prompt_id, _title, _created_at, + _response_type, _response_value, _responded_at, _created_by} -> + execution_id + end, + fn {_execution_id, run_ext_id, input_number, _key, _prompt_id, title, _created_at, + response_type, _response_value, _responded_at, _response_created_by} -> + {input_external_id(run_ext_id, input_number), + {:input, title, if(response_type, do: decode_input_response_type(response_type))}} + end + ) + |> Map.new(fn {execution_id, deps} -> {execution_id, Map.new(deps)} end) + + asset_deps_by_execution = + run_asset_deps + |> Enum.filter(fn {execution_id, _asset_id} -> requested?.(execution_id) end) + |> Enum.group_by( + fn {execution_id, _asset_id} -> execution_id end, + fn {_execution_id, asset_id} -> + {external_id, name, total_count, total_size, entry} = resolve_asset(db, asset_id) + {external_id, {:asset, {name, total_count, total_size, entry}}} + end + ) + |> Map.new(fn {execution_id, deps} -> {execution_id, Map.new(deps)} end) + + metric_definitions_by_execution = + run_metric_defs + |> Enum.filter(fn row -> requested?.(elem(row, 0)) end) + |> Enum.group_by( + fn {execution_id, _, _, _, _, _, _, _, _, _, _} -> execution_id end, + fn {_, key, group, group_units, group_lower, group_upper, scale, units, progress, + lower, upper} -> + {key, + %{ + group: group, + group_units: group_units, + group_lower: group_lower, + group_upper: group_upper, + scale: scale, + units: units, + progress: progress == 1, + lower: lower, + upper: upper + }} + end + ) + |> Map.new(fn {execution_id, defs} -> {execution_id, Map.new(defs)} end) + + Map.new(requested, fn {execution_id, workspace_id, step, attempt} -> + {result, result_at, completed_at, result_created_by} = + case Results.get_result(db, execution_id) do + {:ok, {result, result_at, completion_at, created_by}} -> + {build_result(result, db), result_at, completion_at, created_by} + + {:ok, nil} -> + {nil, nil, nil, nil} + end + + {:ok, asset_ids} = Results.get_assets_for_execution(db, execution_id) + + assets = + asset_ids + |> Enum.map(&resolve_asset(db, &1)) + |> Map.new(fn {external_id, name, total_count, total_size, entry} -> + {external_id, {name, total_count, total_size, entry}} + end) + + result_deps = + run_dependencies + |> Map.get(execution_id, []) + |> Map.new(fn dependency_ref_id -> + {ext_id, _module, _target} = + execution = resolve_execution_ref(db, dependency_ref_id) + + {ext_id, {:result, execution}} + end) + + stream_deps = + run_stream_dependencies + |> Map.get(execution_id, []) + |> Map.new(fn stream_ref_id -> + {:ok, {stream_run_ext_id, step_number, index, module, target}} = + Streams.get_stream_ref(db, stream_ref_id) + + id = stream_external_id(stream_run_ext_id, step_number, index) + {id, {:stream, id, module, target}} + end) + + dependencies = + [ + build_argument_dependencies(db, step.id, step.wait_for), + result_deps, + stream_deps, + Map.get(input_deps_by_execution, execution_id, %{}), + Map.get(asset_deps_by_execution, execution_id, %{}) + ] + |> Enum.reduce(%{}, &Map.merge(&2, &1)) + + # Nothing is outstanding for an execution that has finished: it + # isn't waiting on anything any more, whatever state its + # dependencies are in. + 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, + execution_id, + step.id, + Map.fetch!(workspace_chains, workspace_id), + attempt + ) + + {execution_external_id(run.external_id, step.number, attempt), + %{ + assets: assets, + dependencies: dependencies, + pending_dependencies: pending_dependencies, + inputs: Map.get(submitted_inputs_by_execution, execution_id, %{}), + result: result, + result_at: result_at, + result_created_by: result_created_by, + metric_definitions: Map.get(metric_definitions_by_execution, execution_id, %{}), + checkpoints: %{ + before: enrich_checkpoints(checkpoints_before, db), + after: enrich_checkpoints(checkpoints_after, db) + } + }} + end) + end + + requested_steps = + Enum.flat_map(step_numbers, fn number -> + case Map.fetch(steps_by_number, number) do + {:ok, step} -> [step] + :error -> [] + end + end) + + streams_by_step = + if requested_steps == [] do + %{} + else + {:ok, run_streams} = Streams.get_streams_for_run(db, run.id) + step_ids = MapSet.new(requested_steps, & &1.id) + + run_streams + |> Enum.filter(&MapSet.member?(step_ids, &1.step_id)) + |> then(&build_run_streams(db, &1)) + end + + step_details = + Map.new(requested_steps, fn step -> + {:ok, arguments} = Runs.get_step_arguments(db, step.id) + + {step.number, + %{ + arguments: Enum.map(arguments, &build_value(&1, db)), + streams: Map.get(streams_by_step, step.id, %{}) + }} + end) + + %{executions: executions, steps: step_details} + end + + defp get_tag_set(_db, nil), do: %{} + + defp get_tag_set(db, tag_set_id) do + case TagSets.get_tag_set(db, tag_set_id) do + {:ok, tag_set} -> tag_set + end + end + + defp load_cache_configs(db, steps) do + steps + |> Enum.map(& &1.cache_config_id) + |> Enum.reject(&is_nil/1) + |> Enum.uniq() + |> Map.new(fn cache_config_id -> + case CacheConfigs.get_cache_config(db, cache_config_id) do + {:ok, cache_config} -> {cache_config_id, cache_config} + end + end) + end + + defp build_successor(nil), do: nil + + defp build_successor({run_ext, step_number, attempt}) do + %{type: "execution", id: execution_external_id(run_ext, step_number, attempt)} + end + + defp build_created_by(nil, nil), do: nil + defp build_created_by(user_ext_id, nil), do: %{type: "user", external_id: user_ext_id} + defp build_created_by(nil, token_ext_id), do: %{type: "token", external_id: token_ext_id} + defp ensure_run_in_active_epoch(state, run_external_id) do case Runs.get_run_by_external_id(state.db, run_external_id) do {:ok, run} when not is_nil(run) -> diff --git a/server/lib/coflux/run_view.ex b/server/lib/coflux/run_view.ex new file mode 100644 index 00000000..0ceca405 --- /dev/null +++ b/server/lib/coflux/run_view.ex @@ -0,0 +1,1379 @@ +defmodule Coflux.RunView do + @moduledoc """ + A run's structure, and a projection of it for the run topics. + + The structure is what's needed to decide what to show: steps, their + attempts (with status), the child links from executions to steps, and + groups. The detail of each execution (result, dependencies, assets, and + so on) and of each step (arguments, streams) is held alongside, but only + once loaded: a snapshot carries structure alone, and detail is fetched + for what the projection shows (`missing_details/3`, `put_details/2`). + The projection follows one rule: from the root + step, expand one attempt per step (pinned, else the latest), show that + attempt's ungrouped children and one member of each group (pinned, else + the first), and summarise the rest of each group by branch status. A step + only reachable through an attempt that isn't expanded isn't shown. + + Status roll-ups are kept incrementally so the cost of a notification is + proportional to the depth of the tree rather than the size of the run: + `branch` memoises each step's branch status, and `child_counts` and + `group_counts` count the branch statuses of an execution's children so a + step's status can be recomputed without walking them. + + `apply/2` returns the updated view with the *effects* of the notification: + which steps' projected entries may have changed (`dirty`), which steps' + branch status changed (`branches`), and the structural `events` (a new + step, a new attempt, a new child link) that can change which steps are + shown. + """ + + alias Coflux.RunView.Format + + @branch_priority [:running, :assigning, :errored, :aborted, :suspended] + + @empty_pins %{steps: %{}, groups: %{}} + @empty_effects %{dirty: MapSet.new(), branches: MapSet.new(), events: []} + + defstruct run: nil, + workspace_ids: [], + initial: nil, + root: :initial, + pins: @empty_pins, + steps: %{}, + executions: %{}, + attempts: %{}, + children: %{}, + child_index: %{}, + group_first: %{}, + parents: %{}, + branch: %{}, + child_counts: %{}, + group_counts: %{}, + input_status: %{}, + input_submissions: %{}, + input_dependents: %{} + + # --------------------------------------------------------------------------- + # Building + + @doc """ + Builds a view from a run snapshot (the `steps` map `subscribe_run` + returns). Executions outside `workspace_ids` are left out, as are child + links from them; steps are kept whatever workspace their attempts are in, + since a later attempt may be in a shown workspace. A snapshot may carry + detail (a step's `arguments`, an execution's `result` and so on) or + structure only; what's missing is reported by `missing_details/3`. + """ + def new(run, steps, workspace_ids, opts \\ []) do + view = %__MODULE__{ + run: run, + workspace_ids: workspace_ids, + root: Keyword.get(opts, :root, :initial), + pins: Keyword.get(opts, :pins, @empty_pins) + } + + view = + Enum.reduce(steps, view, fn {number, step}, view -> + put_step(view, number, snapshot_step(number, step)) + end) + + view = + Enum.reduce(steps, view, fn {number, step}, view -> + Enum.reduce(step.executions, view, fn {attempt, execution}, view -> + execution = snapshot_execution(number, attempt, execution) + + if execution.workspace_id in workspace_ids, + do: insert_execution(view, execution), + else: view + end) + end) + + view = + Enum.reduce(steps, view, fn {_number, step}, view -> + Enum.reduce(step.executions, view, fn {_attempt, execution}, view -> + if Map.has_key?(view.executions, execution.execution_id) do + Enum.reduce(execution.children, view, fn {step, attempt, group_id}, view -> + insert_link(view, execution.execution_id, step, attempt, group_id) + end) + else + view + end + end) + end) + + view + |> Map.put(:initial, find_initial(view)) + |> init_branches() + |> init_counts() + |> index_inputs() + end + + defp snapshot_step(number, step) do + %{ + number: number, + module: step.module, + target: step.target, + type: step.type, + parent_id: step.parent_id, + cache_config: step.cache_config, + cache_key: step.cache_key, + memo_key: step.memo_key, + concurrency_key: step.concurrency_key, + concurrency_limit: step.concurrency_limit, + group_key: step.group_key, + group_limit: step.group_limit, + retries: snapshot_retries(step), + recurrent: step.recurrent == 1 or step.recurrent == true, + timeout: step.timeout, + created_at: step.created_at, + arguments: Map.get(step, :arguments), + requires: step.requires, + streams: snapshot_streams(Map.get(step, :streams)) + } + end + + defp snapshot_streams(nil), do: nil + + defp snapshot_streams(streams) do + Map.new(streams, fn {index, stream} -> + {index, + %{ + id: stream.id, + index: stream.index, + position: stream.position, + workspace_id: stream.workspace_id, + buffer: stream.buffer, + timeout_ms: stream.timeout_ms, + opened_at: stream.opened_at, + attempts: stream.attempts, + closed_at: stream.closed_at, + closed_by: stream.closed_by, + reason: if(stream.reason, do: Atom.to_string(stream.reason)), + error: Format.stream_error(stream.error) + }} + end) + end + + defp snapshot_retries(%{retry_limit: 0}), do: nil + + defp snapshot_retries(step) do + %{ + limit: if(step.retry_limit == -1, do: nil, else: step.retry_limit), + backoff_min: step.retry_backoff_min, + backoff_max: step.retry_backoff_max + } + end + + defp snapshot_execution(step, attempt, execution) do + base = %{ + id: execution.execution_id, + step: step, + attempt: attempt, + workspace_id: execution.workspace_id, + created_at: execution.created_at, + created_by: execution.created_by, + execute_after: execution.execute_after, + assigned_at: execution.assigned_at, + completed_at: execution.completed_at, + completion: execution.completion, + groups: execution.groups, + loaded: false, + assets: %{}, + dependencies: %{}, + pending: MapSet.new(), + inputs: %{}, + result: nil, + result_at: nil, + result_created_by: nil, + inner_result: nil, + metrics: %{}, + checkpoints: %{before: %{}, after: %{}} + } + + if Map.has_key?(execution, :result), do: apply_detail(base, execution), else: base + end + + # Detail as `build_run_details` shapes it, replacing whatever the + # execution held: the database is authoritative at the moment it was + # read, and any notification still queued re-applies on top. + defp apply_detail(execution, detail) do + %{ + execution + | loaded: true, + assets: detail.assets, + dependencies: detail.dependencies, + pending: Map.get(detail, :pending_dependencies, MapSet.new()), + inputs: Map.get(detail, :inputs, %{}), + result: detail.result, + result_at: detail.result_at, + result_created_by: detail.result_created_by, + inner_result: nil, + metrics: detail.metric_definitions, + checkpoints: detail.checkpoints + } + end + + defp find_initial(view) do + view.steps + |> Map.values() + |> Enum.reject(& &1.parent_id) + |> Enum.min_by(& &1.created_at, fn -> nil end) + |> case do + nil -> nil + step -> step.number + end + end + + defp init_branches(view) do + Enum.reduce(Map.keys(view.steps), view, fn step, view -> + {view, _status} = ensure_branch(view, step, MapSet.new()) + view + end) + end + + # Memoised, bottom-up. A step reached again while it's being computed (a + # cycle through memo links) contributes nothing to its own status. + defp ensure_branch(view, step, visiting) do + case Map.fetch(view.branch, step) do + {:ok, status} -> + {view, status} + + :error -> + cond do + MapSet.member?(visiting, step) -> + {view, nil} + + true -> + case latest_execution(view, step) do + nil -> + {view, nil} + + execution -> + visiting = MapSet.put(visiting, step) + + {view, statuses} = + Enum.reduce(child_steps(view, execution.id), {view, []}, fn child, + {view, statuses} -> + {view, status} = ensure_branch(view, child, visiting) + {view, [status | statuses]} + end) + + status = combine(own_status(execution), statuses) + {%{view | branch: Map.put(view.branch, step, status)}, status} + end + end + end + end + + defp init_counts(view) do + Enum.reduce(view.child_index, view, fn {execution_id, children}, view -> + Enum.reduce(children, view, fn {step, group_id}, view -> + move_count(view, execution_id, group_id, nil, Map.get(view.branch, step)) + end) + end) + end + + defp index_inputs(view) do + Enum.reduce(view.executions, view, fn {_id, execution}, view -> + index_execution_inputs(view, execution) + end) + end + + defp index_execution_inputs(view, execution) do + view = + Enum.reduce(execution.inputs, view, fn {input_id, input}, view -> + view + |> index_input(:input_submissions, input_id, execution.id) + |> record_input_status(input_id, input.status) + end) + + Enum.reduce(execution.dependencies, view, fn + {input_id, {:input, _title, status}}, view -> + view + |> index_input(:input_dependents, input_id, execution.id) + |> record_input_status(input_id, status) + + _other, view -> + view + end) + end + + # --------------------------------------------------------------------------- + # Detail, loaded for what's shown + + @doc """ + What `build_run_details` needs to be asked for before the given steps can + be projected: each step whose arguments aren't loaded, and (unless + `executions?` is false) each step's expanded execution if its detail + isn't loaded. Shaped as the request `Orchestration.get_run_details` takes. + """ + def missing_details(view, steps, executions? \\ true) do + Enum.reduce(steps, %{executions: [], steps: []}, fn number, request -> + request = + case Map.get(view.steps, number) do + %{arguments: nil} -> %{request | steps: [number | request.steps]} + _ -> request + end + + case executions? && expanded_execution(view, number) do + %{loaded: false, attempt: attempt} -> + %{request | executions: [{number, attempt} | request.executions]} + + _ -> + request + end + end) + end + + def empty_request?(%{executions: [], steps: []}), do: true + def empty_request?(_request), do: false + + @doc "Merges a `get_run_details` reply into the view." + def put_details(view, %{executions: executions, steps: steps}) do + view = + Enum.reduce(executions, view, fn {execution_id, detail}, view -> + case Map.fetch(view.executions, execution_id) do + {:ok, execution} -> + execution = apply_detail(execution, detail) + view |> store(execution) |> index_execution_inputs(execution) + + :error -> + view + end + end) + + Enum.reduce(steps, view, fn {number, detail}, view -> + case Map.fetch(view.steps, number) do + {:ok, step} -> + put_step(view, number, %{ + step + | arguments: detail.arguments, + streams: snapshot_streams(detail.streams) + }) + + :error -> + view + end + end) + end + + # --------------------------------------------------------------------------- + # Structure primitives (no effects, no roll-ups) + + defp put_step(view, number, step) do + %{view | steps: Map.put(view.steps, number, step)} + end + + defp insert_execution(view, execution) do + %{ + view + | executions: Map.put(view.executions, execution.id, execution), + attempts: + Map.update( + view.attempts, + execution.step, + %{execution.attempt => execution.id}, + &Map.put(&1, execution.attempt, execution.id) + ) + } + end + + # A link is kept once per (parent execution, step): a parent that + # re-submits a step it already links to (a memo hit on the same target) + # doesn't get a second entry. + defp insert_link(view, parent_id, step, attempt, group_id) do + index = Map.get(view.child_index, parent_id, %{}) + + if Map.has_key?(index, step) do + view + else + link = %{step: step, attempt: attempt, group_id: group_id} + + group_first = + if group_id, + do: Map.put_new(view.group_first, {parent_id, group_id}, step), + else: view.group_first + + %{ + view + | children: Map.update(view.children, parent_id, [link], &(&1 ++ [link])), + child_index: Map.put(view.child_index, parent_id, Map.put(index, step, group_id)), + parents: Map.update(view.parents, step, [parent_id], &(&1 ++ [parent_id])), + group_first: group_first + } + end + end + + defp index_input(view, field, input_id, execution_id) do + Map.update!(view, field, fn index -> + Map.update(index, input_id, MapSet.new([execution_id]), &MapSet.put(&1, execution_id)) + end) + end + + defp record_input_status(view, _input_id, nil), do: view + + defp record_input_status(view, input_id, status) do + %{view | input_status: Map.put(view.input_status, input_id, status)} + end + + # --------------------------------------------------------------------------- + # Status + + defp own_status(%{completion: nil, assigned_at: nil}), do: :assigning + defp own_status(%{completion: nil}), do: :running + defp own_status(%{completion: %{kind: kind}}), do: completion_status(kind) + + defp completion_status(kind) + when kind in ["succeeded", "recurred", "stream_errored", "stream_timeout"], + do: :completed + + defp completion_status("errored"), do: :errored + + defp completion_status(kind) when kind in ["abandoned", "cancelled", "crashed", "timeout"], + do: :aborted + + defp completion_status("suspended"), do: :suspended + defp completion_status(kind) when kind in ["deferred", "cached", "spawned"], do: :deferred + + # The roll-up the graph shows for a branch: anything still going wins, + # then anything that went wrong; otherwise it's done. + defp combine(own, child_statuses) do + Enum.find(@branch_priority, :completed, fn status -> + status == own or status in child_statuses + end) + end + + defp compute_branch(view, step) do + case latest_execution(view, step) do + nil -> + nil + + execution -> + statuses = + view.child_counts + |> Map.get(execution.id, %{}) + |> Enum.filter(fn {_status, count} -> count > 0 end) + |> Enum.map(&elem(&1, 0)) + + combine(own_status(execution), statuses) + end + end + + defp move_count(view, parent_id, group_id, from, to) do + view = %{view | child_counts: shift_in(view.child_counts, parent_id, from, to)} + + if group_id do + %{view | group_counts: shift_in(view.group_counts, {parent_id, group_id}, from, to)} + else + view + end + end + + defp shift_in(counts, key, from, to) do + Map.put(counts, key, shift(Map.get(counts, key, %{}), from, to)) + end + + defp shift(counts, from, to) do + counts + |> then(fn counts -> if from, do: Map.update(counts, from, 0, &(&1 - 1)), else: counts end) + |> then(fn counts -> if to, do: Map.update(counts, to, 1, &(&1 + 1)), else: counts end) + end + + # Recomputes a step's branch status and, if it changed, moves the counts + # on every parent execution that links to it and continues upward from + # each parent that is its step's latest attempt. + defp recompute_branch(view, step, effects, visited) do + new = compute_branch(view, step) + old = Map.get(view.branch, step) + + if new == old do + {view, effects} + else + view = %{view | branch: Map.put(view.branch, step, new)} + effects = %{effects | branches: MapSet.put(effects.branches, step)} + visited = MapSet.put(visited, step) + + Enum.reduce(Map.get(view.parents, step, []), {view, effects}, fn parent_id, + {view, effects} -> + group_id = view.child_index[parent_id][step] + view = move_count(view, parent_id, group_id, old, new) + parent = view.executions[parent_id] + effects = dirty(effects, parent.step) + + if latest(view, parent.step) == parent.attempt and + not MapSet.member?(visited, parent.step) do + recompute_branch(view, parent.step, effects, visited) + else + {view, effects} + end + end) + end + end + + defp status_changed(view, execution_id, effects) do + execution = view.executions[execution_id] + effects = dirty(effects, execution.step) + + if latest(view, execution.step) == execution.attempt do + recompute_branch(view, execution.step, effects, MapSet.new()) + else + {view, effects} + end + end + + # --------------------------------------------------------------------------- + # Notifications + + def apply_all(view, notifications) do + {view, effects} = + Enum.reduce(notifications, {view, @empty_effects}, fn notification, {view, effects} -> + {view, more} = __MODULE__.apply(view, notification) + {view, merge_effects(effects, more)} + end) + + {view, %{effects | events: Enum.reverse(effects.events)}} + end + + defp merge_effects(a, b) do + %{ + dirty: MapSet.union(a.dirty, b.dirty), + branches: MapSet.union(a.branches, b.branches), + events: b.events ++ a.events + } + end + + defp dirty(effects, step), do: %{effects | dirty: MapSet.put(effects.dirty, step)} + defp event(effects, event), do: %{effects | events: [event | effects.events]} + + def apply(view, {:step, number, step, _workspace_external_id}) do + if Map.has_key?(view.steps, number) do + {view, @empty_effects} + else + step = %{ + number: number, + module: step.module, + target: step.target, + type: step.type, + parent_id: step.parent_id, + cache_config: step.cache_config, + cache_key: step.cache_key, + memo_key: step.memo_key, + concurrency_key: step.concurrency_key, + concurrency_limit: step.concurrency_limit, + group_key: step.group_key, + group_limit: step.group_limit, + retries: step.retries, + recurrent: step.recurrent == true, + timeout: step.timeout, + created_at: step.created_at, + arguments: step.arguments, + requires: step.requires, + streams: %{} + } + + {put_step(view, number, step), event(@empty_effects, {:step, number})} + end + end + + def apply( + view, + {:execution, step, attempt, execution_id, workspace_id, created_at, execute_after, + dependencies, created_by, checkpoints, pending} + ) do + execution = %{ + id: execution_id, + step: step, + attempt: attempt, + workspace_id: workspace_id, + created_at: created_at, + created_by: created_by, + execute_after: execute_after, + assigned_at: nil, + result_at: nil, + completed_at: nil, + completion: nil, + groups: %{}, + loaded: true, + assets: %{}, + dependencies: + Map.new(dependencies, fn {id, dependency} -> {id, tag_result(dependency)} end), + pending: pending, + inputs: %{}, + result: nil, + result_created_by: nil, + inner_result: nil, + metrics: %{}, + # Nothing has run yet, so what the execution will start from is also + # what it currently holds. + checkpoints: %{before: checkpoints, after: checkpoints} + } + + cond do + execution.workspace_id not in view.workspace_ids -> + {view, @empty_effects} + + not Map.has_key?(view.steps, step) -> + {view, @empty_effects} + + Map.has_key?(view.executions, execution_id) -> + {view, @empty_effects} + + true -> + # Reported whether or not the step had an attempt before: a first + # attempt in a shown workspace can bring an already-linked step + # into view, a later one can change which subtree is shown. + view = insert_execution(view, execution) + effects = @empty_effects |> dirty(step) |> event({:attempt, step}) + + if latest(view, step) == attempt do + recompute_branch(view, step, effects, MapSet.new()) + else + {view, effects} + end + end + end + + def apply(view, {:child, parent_id, {step, attempt, group_id}}) do + cond do + not Map.has_key?(view.executions, parent_id) -> + {view, @empty_effects} + + not Map.has_key?(view.steps, step) -> + {view, @empty_effects} + + Map.has_key?(Map.get(view.child_index, parent_id, %{}), step) -> + {view, @empty_effects} + + true -> + view = insert_link(view, parent_id, step, attempt, group_id) + parent = view.executions[parent_id] + + effects = + @empty_effects + |> dirty(parent.step) + |> event({:link, parent_id, step, group_id}) + + case Map.get(view.branch, step) do + nil -> + {view, effects} + + status -> + view = move_count(view, parent_id, group_id, nil, status) + + if latest(view, parent.step) == parent.attempt do + recompute_branch(view, parent.step, effects, MapSet.new()) + else + {view, effects} + end + end + end + end + + def apply(view, {:assigned, assigned}) do + Enum.reduce(assigned, {view, @empty_effects}, fn {execution_id, assigned_at}, + {view, effects} -> + case Map.fetch(view.executions, execution_id) do + {:ok, execution} -> + view = store(view, %{execution | assigned_at: assigned_at}) + {view, more} = status_changed(view, execution_id, @empty_effects) + {view, merge_effects(effects, more)} + + :error -> + {view, effects} + end + end) + end + + def apply(view, {:completion, execution_id, kind, successor, completed_at}) do + case Map.fetch(view.executions, execution_id) do + {:ok, execution} -> + # Nothing is outstanding once the execution has finished, whatever + # state its dependencies are in - the rule the snapshot applies too. + view = + store(view, %{ + execution + | completed_at: completed_at, + completion: %{kind: Atom.to_string(kind), successor: successor}, + pending: MapSet.new() + }) + + status_changed(view, execution_id, @empty_effects) + + :error -> + {view, @empty_effects} + end + end + + def apply(view, {:result, execution_id, result, result_at, created_by}) do + update_execution(view, execution_id, fn execution -> + %{execution | result: result, result_at: result_at, result_created_by: created_by} + end) + end + + # The result of a deferred/cached/spawned execution's target, shown nested + # inside the redirecting result. + def apply(view, {:result_result, execution_id, result, _created_at, created_by}) do + update_execution(view, execution_id, fn execution -> + %{execution | inner_result: {result, created_by}} + end) + end + + def apply(view, {:metric_defined, execution_id, key, definition}) do + update_execution(view, execution_id, fn execution -> + metric = %{ + group: Map.get(definition, "group"), + group_units: Map.get(definition, "group_units"), + group_lower: Map.get(definition, "group_lower"), + group_upper: Map.get(definition, "group_upper"), + scale: Map.get(definition, "scale"), + units: Map.get(definition, "units"), + progress: Map.get(definition, "progress", false), + lower: Map.get(definition, "lower"), + upper: Map.get(definition, "upper") + } + + %{execution | metrics: Map.put(execution.metrics, key, metric)} + end) + end + + def apply(view, {:group, execution_id, group_id, name, concurrency}) do + update_execution(view, execution_id, fn execution -> + %{ + execution + | groups: Map.put(execution.groups, group_id, %{name: name, concurrency: concurrency}) + } + end) + end + + def apply(view, {:asset, execution_id, asset_id, asset}) do + update_execution(view, execution_id, fn execution -> + %{execution | assets: Map.put(execution.assets, asset_id, asset)} + 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. + def apply(view, {:pending_dependencies, execution_id, pending}) do + update_execution(view, execution_id, fn execution -> %{execution | pending: pending} end) + end + + def apply(view, {:result_dependency, execution_id, dependency_id, dependency, pending}) do + update_execution(view, execution_id, fn execution -> + execution + |> put_dependency(dependency_id, tag_result(dependency)) + |> set_pending(dependency_id, pending) + end) + end + + def apply(view, {:stream_dependency, execution_id, stream_id, module, target, pending}) do + update_execution(view, execution_id, fn execution -> + execution + |> put_dependency(stream_id, {:stream, stream_id, module, target}) + |> set_pending(stream_id, pending) + end) + end + + def apply(view, {:asset_dependency, execution_id, asset_id, asset, pending}) do + update_execution(view, execution_id, fn execution -> + execution + |> put_dependency(asset_id, {:asset, asset}) + |> set_pending(asset_id, pending) + end) + end + + def apply(view, {:input_dependency, execution_id, input_id, title, response_type, pending}) do + {view, effects} = + update_execution(view, execution_id, fn execution -> + execution + |> put_dependency(input_id, {:input, title, response_type}) + |> set_pending(input_id, pending) + end) + + view = + if Map.has_key?(view.executions, execution_id) do + view + |> index_input(:input_dependents, input_id, execution_id) + |> record_input_status(input_id, response_type) + else + view + end + + {view, effects} + end + + def apply(view, {:input_submitted, execution_id, input_id, title}) do + status = Map.get(view.input_status, input_id) + + {view, effects} = + update_execution(view, execution_id, fn execution -> + %{ + execution + | inputs: Map.put(execution.inputs, input_id, %{title: title, status: status}) + } + end) + + view = + if Map.has_key?(view.executions, execution_id), + do: index_input(view, :input_submissions, input_id, execution_id), + else: view + + {view, effects} + end + + def apply(view, {:input_response, input_id, response_type}) do + view = record_input_status(view, input_id, response_type) + + {view, effects} = + view.input_submissions + |> Map.get(input_id, MapSet.new()) + |> Enum.reduce({view, @empty_effects}, fn execution_id, {view, effects} -> + {view, more} = + update_execution(view, execution_id, fn execution -> + %{ + execution + | inputs: + Map.update!(execution.inputs, input_id, &Map.put(&1, :status, response_type)) + } + end) + + {view, merge_effects(effects, more)} + end) + + view.input_dependents + |> Map.get(input_id, MapSet.new()) + |> Enum.reduce({view, effects}, fn execution_id, {view, effects} -> + {view, more} = + update_execution(view, execution_id, fn execution -> + case Map.fetch(execution.dependencies, input_id) do + {:ok, {:input, title, _status}} -> + put_dependency(execution, input_id, {:input, title, response_type}) + + _ -> + execution + end + end) + + {view, merge_effects(effects, more)} + end) + end + + def apply(view, {:checkpoints, execution_id, checkpoints}) do + # Only the "after" side moves — what the execution started from was + # fixed when it was created. + update_execution(view, execution_id, fn execution -> + %{execution | checkpoints: %{execution.checkpoints | after: checkpoints}} + end) + end + + # An execution registered on one of the step's streams — opening it, or + # resuming it after a suspend (`continued`). + def apply(view, {:stream_registered, step, index, info}) do + cond do + info.workspace_id not in view.workspace_ids -> + {view, @empty_effects} + + not Map.has_key?(view.steps, step) -> + {view, @empty_effects} + + true -> + update_stream(view, step, index, fn + nil -> + %{ + id: info.id, + index: index, + position: info.position, + workspace_id: info.workspace_id, + buffer: info.buffer, + timeout_ms: info.timeout_ms, + opened_at: info.opened_at, + attempts: [info.attempt], + closed_at: nil, + closed_by: nil, + reason: nil, + error: nil + } + + stream -> + attempts = + if info.attempt in stream.attempts, + do: stream.attempts, + else: stream.attempts ++ [info.attempt] + + %{stream | attempts: attempts, buffer: info.buffer, timeout_ms: info.timeout_ms} + end) + end + end + + def apply(view, {:stream_closed, step, index, reason, error, attempt, closed_at}) do + if get_in(view.steps, [step, :streams]) && get_in(view.steps, [step, :streams, index]) do + update_stream(view, step, index, fn stream -> + %{stream | closed_at: closed_at, closed_by: attempt, reason: reason, error: error} + end) + else + {view, @empty_effects} + end + end + + defp update_execution(view, execution_id, fun) do + case Map.fetch(view.executions, execution_id) do + {:ok, execution} -> + {store(view, fun.(execution)), dirty(@empty_effects, execution.step)} + + :error -> + {view, @empty_effects} + end + end + + defp update_stream(view, step, index, fun) do + view = + update_in(view.steps[step].streams, fn streams -> + streams = streams || %{} + Map.put(streams, index, fun.(Map.get(streams, index))) + end) + + {view, dirty(@empty_effects, step)} + end + + defp store(view, execution) do + %{view | executions: Map.put(view.executions, execution.id, execution)} + end + + defp put_dependency(execution, id, dependency) do + %{execution | dependencies: Map.put(execution.dependencies, id, dependency)} + end + + defp set_pending(execution, id, true), + do: %{execution | pending: MapSet.put(execution.pending, id)} + + defp set_pending(execution, id, _), + do: %{execution | pending: MapSet.delete(execution.pending, id)} + + # Result dependencies arrive both tagged (from the snapshot, and argument + # dependencies) and as a bare execution reference (from notifications). + defp tag_result({:result, _execution} = dependency), do: dependency + defp tag_result({_ext_id, _module, _target} = execution), do: {:result, execution} + + # --------------------------------------------------------------------------- + # Reading the structure + + def latest(view, step) do + case Map.get(view.attempts, step) do + nil -> nil + attempts when map_size(attempts) == 0 -> nil + attempts -> attempts |> Map.keys() |> Enum.max() + end + end + + defp latest_execution(view, step) do + case latest(view, step) do + nil -> nil + attempt -> view.executions[view.attempts[step][attempt]] + end + end + + @doc "The attempt of `step` that's expanded: the pinned one, else the latest." + def expanded(view, step) do + case Map.get(view.pins.steps, step) do + nil -> latest(view, step) + attempt -> if Map.has_key?(Map.get(view.attempts, step, %{}), attempt), do: attempt + end + end + + def expanded_execution(view, step) do + case expanded(view, step) do + nil -> nil + attempt -> view.executions[view.attempts[step][attempt]] + end + end + + def branch_status(view, step), do: Map.get(view.branch, step) + + defp child_steps(view, execution_id) do + view.child_index |> Map.get(execution_id, %{}) |> Map.keys() + end + + @doc "The step shown for a group: the pinned member, else the first." + def chosen_member(view, execution_id, group_id) do + Map.get(view.pins.groups, {execution_id, group_id}) || + Map.get(view.group_first, {execution_id, group_id}) + end + + @doc "The child links of an execution that the projection shows." + def visible_children(view, execution_id) do + view.children + |> Map.get(execution_id, []) + |> Enum.filter(fn + %{group_id: nil} -> true + %{group_id: group_id, step: step} -> chosen_member(view, execution_id, group_id) == step + end) + end + + @doc "Whether a link from `parent_id` to `step` is one the projection shows." + def link_visible?(view, parent_id, step, group_id) do + case Map.fetch(view.executions, parent_id) do + {:ok, parent} -> + expanded(view, parent.step) == parent.attempt and + (is_nil(group_id) or chosen_member(view, parent_id, group_id) == step) + + :error -> + false + end + end + + def root_step(view) do + case view.root do + :initial -> view.initial + :none -> nil + {:step, step} -> step + end + end + + @doc "The steps the projection shows, from the root." + def visible_steps(view) do + case root_step(view) do + nil -> MapSet.new() + step -> reachable(view, step, MapSet.new()) + end + end + + @doc """ + The steps reached from `step` through expanded attempts and shown + children, not counting those in `seen` (or anything only reachable through + them). + """ + def reachable(view, step, seen) do + MapSet.difference(traverse(view, step, seen), seen) + end + + defp traverse(view, step, seen) do + cond do + MapSet.member?(seen, step) -> + seen + + not Map.has_key?(view.steps, step) -> + seen + + true -> + case expanded_execution(view, step) do + nil -> + seen + + execution -> + seen = MapSet.put(seen, step) + + Enum.reduce(visible_children(view, execution.id), seen, fn link, seen -> + traverse(view, link.step, seen) + end) + end + end + end + + # --------------------------------------------------------------------------- + # Selection + + @doc """ + What pinning `execution_id` changes about the default view. Walks up from + the execution (through the most recently created parent at each hop, as + the graph does) and finds the topmost place the pinned path leaves the + default: a pinned attempt that isn't the step's latest, or a pinned group + member that isn't the group's first. + + Returns `nil` for an unknown execution. Otherwise a map with `root` + (`{:step, number}` at which the view diverges, or `:none` when the + execution is in the default view already), `parent` (the hop from the + default view into the root, if any), `pins`, and `path` (the steps on the + pinned path, for deciding whether a change touches it). + """ + def selection(view, execution_id) do + case Map.fetch(view.executions, execution_id) do + :error -> + nil + + {:ok, execution} -> + hops = climb(view, execution.step, [], MapSet.new([execution.step])) + + pins = + Enum.reduce(hops, %{steps: %{execution.step => execution.attempt}, groups: %{}}, fn + {parent_id, group_id, step}, pins -> + parent = view.executions[parent_id] + steps = Map.put(pins.steps, parent.step, parent.attempt) + + groups = + if group_id, + do: Map.put(pins.groups, {parent_id, group_id}, step), + else: pins.groups + + %{steps: steps, groups: groups} + end) + + {root, parent} = divergence(view, hops, execution) + + %{ + root: root, + parent: parent, + pins: pins, + path: MapSet.new(Map.keys(pins.steps)) + } + end + end + + # Hops from the top of the run down to `step`: `{parent execution, group, + # child step}`. Stops at a step with no shown parent, or one already on + # the path (a cycle through memo links). + defp climb(view, step, hops, seen) do + case most_recent_parent(view, step) do + nil -> + hops + + parent_id -> + parent = view.executions[parent_id] + + if MapSet.member?(seen, parent.step) do + hops + else + group_id = view.child_index[parent_id][step] + + climb( + view, + parent.step, + [{parent_id, group_id, step} | hops], + MapSet.put(seen, parent.step) + ) + end + end + end + + defp most_recent_parent(view, step) do + view.parents + |> Map.get(step, []) + |> Enum.max_by(fn parent_id -> view.executions[parent_id].created_at end, fn -> nil end) + end + + defp divergence(view, hops, execution) do + result = + Enum.reduce_while(hops, nil, fn {parent_id, group_id, step}, above -> + parent = view.executions[parent_id] + + cond do + latest(view, parent.step) != parent.attempt -> + {:halt, {:diverged, parent.step, above}} + + group_id != nil and Map.get(view.group_first, {parent_id, group_id}) != step -> + {:halt, {:diverged, step, %{execution_id: parent_id, group_id: group_id}}} + + true -> + {:cont, %{execution_id: parent_id, group_id: group_id}} + end + end) + + case result do + {:diverged, step, parent} -> + {{:step, step}, parent} + + above -> + if latest(view, execution.step) != execution.attempt, + do: {{:step, execution.step}, above}, + else: {:none, nil} + end + end + + def with_selection(view, selection) do + %{view | root: selection.root, pins: selection.pins} + end + + @doc "Whether any of the structural events could change a selection's path." + def touches?(view, events, path) do + Enum.any?(events, fn + {:attempt, step} -> + MapSet.member?(path, step) + + {:link, parent_id, step, _group_id} -> + MapSet.member?(path, step) or MapSet.member?(path, view.executions[parent_id].step) + + _ -> + false + end) + end + + # --------------------------------------------------------------------------- + # Projection + + def step_key(view, step), do: Format.step_key(view.run.external_id, step) + + @doc "The projected `steps` map: every visible step." + def project(view) do + view + |> visible_steps() + |> Map.new(fn step -> {step_key(view, step), project_step(view, step)} end) + end + + def project_step(view, number) do + step = view.steps[number] + attempts = Map.get(view.attempts, number, %{}) + + executions = + case expanded(view, number) do + nil -> + %{} + + attempt -> + %{ + Integer.to_string(attempt) => + project_execution(view, view.executions[attempts[attempt]]) + } + end + + %{ + stepNumber: number, + module: step.module, + target: step.target, + type: step.type, + parentId: step.parent_id, + cacheConfig: Format.cache_config(step.cache_config), + cacheKey: Format.key(step.cache_key), + memoKey: Format.key(step.memo_key), + concurrency: Format.concurrency(step), + group: Format.group(step), + retries: Format.retries(step.retries), + recurrent: step.recurrent, + timeout: step.timeout, + createdAt: step.created_at, + arguments: Enum.map(step.arguments || [], &Format.value/1), + requires: step.requires, + streams: Format.streams(step.streams || %{}, view.workspace_ids), + attempts: + Map.new(attempts, fn {attempt, execution_id} -> + {Integer.to_string(attempt), attempt_summary(view.executions[execution_id])} + end), + executions: executions + } + end + + defp attempt_summary(execution) do + %{ + executionId: execution.id, + workspaceId: execution.workspace_id, + createdAt: execution.created_at, + executeAfter: execution.execute_after, + assignedAt: execution.assigned_at, + completedAt: execution.completed_at, + completion: execution.completion + } + end + + defp project_execution(view, execution) do + result = Format.result(execution.result, execution.result_created_by) + + result = + case {result, execution.inner_result} do + {%{result: _}, {inner, created_by}} -> + %{result | result: Format.result(inner, created_by)} + + _ -> + result + end + + %{ + executionId: execution.id, + workspaceId: execution.workspace_id, + createdAt: execution.created_at, + createdBy: Format.principal(execution.created_by), + executeAfter: execution.execute_after, + assignedAt: execution.assigned_at, + resultAt: execution.result_at, + completedAt: execution.completed_at, + completion: execution.completion, + groups: + Map.new(execution.groups, fn {group_id, group} -> + {Integer.to_string(group_id), + %{ + name: group.name, + concurrency: group.concurrency, + members: Format.members(Map.get(view.group_counts, {execution.id, group_id}, %{})) + }} + end), + assets: + Map.new(execution.assets, fn {asset_id, asset} -> {asset_id, Format.asset(asset)} end), + dependencies: Format.dependencies(execution.dependencies, execution.pending), + children: + view + |> visible_children(execution.id) + |> Enum.map(&Format.child(&1, view.run.external_id)), + inputs: execution.inputs, + result: result, + metrics: Map.new(execution.metrics, fn {key, metric} -> {key, Format.metric(metric)} end), + checkpoints: Format.checkpoints(execution.checkpoints.before, execution.checkpoints.after) + } + end + + @doc """ + Every step with an attempt in a shown workspace, as structure only: the + step's identity, and each attempt summarised with its complete list of + children. Nothing is collapsed and nothing needs detail, so it's what the + whole-run pages (timeline, logs) work from. + """ + def project_structure(view) do + Map.new(view.attempts, fn {number, _attempts} -> + {step_key(view, number), project_structure_step(view, number)} + end) + end + + def project_structure_step(view, number) do + step = view.steps[number] + + %{ + stepNumber: number, + module: step.module, + target: step.target, + type: step.type, + parentId: step.parent_id, + createdAt: step.created_at, + attempts: + Map.new(Map.get(view.attempts, number, %{}), fn {attempt, execution_id} -> + execution = view.executions[execution_id] + + children = + view.children + |> Map.get(execution_id, []) + |> Enum.map(&Format.child(&1, view.run.external_id)) + + {Integer.to_string(attempt), Map.put(attempt_summary(execution), :children, children)} + end) + } + end + + @doc "The members of a group, in the order they were linked, with branch status." + def group_members(view, execution_id, group_id) do + view.children + |> Map.get(execution_id, []) + |> Enum.filter(&(&1.group_id == group_id)) + |> Enum.map(&group_member(view, &1)) + end + + # A member is listed with its latest attempt (what selecting it opens) + # rather than the attempt it was first linked as, and with its arguments, + # which is how the popover labels it. + def group_member(view, link) do + step = view.steps[link.step] + + %{ + stepId: step_key(view, link.step), + attempt: latest(view, link.step), + module: step.module, + target: step.target, + arguments: Enum.map(step.arguments || [], &Format.value/1), + status: member_status(view, link.step), + createdAt: step.created_at + } + end + + def member_status(view, step) do + case Map.get(view.branch, step) do + nil -> nil + status -> Format.branch_status(status) + end + end +end diff --git a/server/lib/coflux/run_view/diff.ex b/server/lib/coflux/run_view/diff.ex new file mode 100644 index 00000000..1b086582 --- /dev/null +++ b/server/lib/coflux/run_view/diff.ex @@ -0,0 +1,36 @@ +defmodule Coflux.RunView.Diff do + @moduledoc """ + Emits the topic updates that turn one projected value into another. + + Maps are walked key by key so a change deep in a step's entry becomes one + `set` at that path; anything else that differs is set wholesale. Structs + never appear in projected values, but are treated as leaves just in case. + """ + + alias Topical.Topic + + def apply(topic, path, old, new) do + diff(topic, path, old, new) + end + + defp diff(topic, path, old, new) + when is_map(old) and is_map(new) and not is_struct(old) and not is_struct(new) do + topic = + old + |> Map.keys() + |> Enum.reject(&Map.has_key?(new, &1)) + |> Enum.reduce(topic, fn key, topic -> Topic.unset(topic, path, key) end) + + Enum.reduce(new, topic, fn {key, value}, topic -> + case Map.fetch(old, key) do + {:ok, ^value} -> topic + {:ok, previous} -> diff(topic, path ++ [key], previous, value) + :error -> Topic.set(topic, path ++ [key], value) + end + end) + end + + defp diff(topic, path, _old, new) do + Topic.set(topic, path, new) + end +end diff --git a/server/lib/coflux/run_view/format.ex b/server/lib/coflux/run_view/format.ex new file mode 100644 index 00000000..91f29e15 --- /dev/null +++ b/server/lib/coflux/run_view/format.ex @@ -0,0 +1,273 @@ +defmodule Coflux.RunView.Format do + @moduledoc """ + Converts what a `Coflux.RunView` holds (orchestration's tuples and maps) + into the shape the run topics publish. + """ + + import Coflux.TopicUtils + + @branch_statuses [:assigning, :running, :completed, :errored, :aborted, :suspended] + + def step_key(run_external_id, step_number), do: "#{run_external_id}:#{step_number}" + + def child(%{step: step, attempt: attempt, group_id: group_id}, run_external_id) do + %{stepId: step_key(run_external_id, step), attempt: attempt, groupId: group_id} + end + + def value(value), do: build_value(value) + + def values(values), do: Map.new(values, fn {name, value} -> {name, build_value(value)} end) + + def principal(principal), do: build_principal(principal) + + def asset(asset), do: build_asset(asset) + + def branch_status(status), do: Atom.to_string(status) + + @doc """ + The summary of a group's members: how many there are, and how many are + in each branch status. Every status is present so the shape is stable. + """ + def members(counts) do + by_status = Map.new(@branch_statuses, fn status -> {status, Map.get(counts, status, 0)} end) + %{total: Enum.sum(Map.values(by_status)), byStatus: by_status} + end + + def cache_config(nil), do: nil + + def cache_config(cache_config) do + %{ + params: cache_config.params, + maxAge: cache_config.max_age, + namespace: cache_config.namespace, + version: cache_config.version + } + end + + def concurrency(%{concurrency_key: nil}), do: nil + + def concurrency(step) do + %{limit: step.concurrency_limit, key: key(step.concurrency_key)} + end + + # The group limit the step counts against, if any. The key is readable + # as it stands ("/"), so no hex prefix. + def group(%{group_key: nil}), do: nil + + def group(step) do + %{limit: step.group_limit, key: step.group_key} + end + + def retries(nil), do: nil + + def retries(%{limit: limit, backoff_min: backoff_min, backoff_max: backoff_max}) do + %{limit: limit, backoffMin: backoff_min, backoffMax: backoff_max} + end + + def key(key, length \\ 10) + def key(nil, _length), do: nil + + def key(key, length) do + key + |> Base.encode16(case: :lower) + |> String.slice(0, length) + end + + def metric(definition) do + %{ + group: definition.group, + groupUnits: definition.group_units, + groupLower: definition.group_lower, + groupUpper: definition.group_upper, + scale: definition.scale, + units: definition.units, + progress: definition.progress, + lower: definition.lower, + upper: definition.upper + } + end + + # `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. + def dependencies(dependencies, pending) do + Map.new(dependencies, fn + {id, {:result, 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, + pending: MapSet.member?(pending, id) + }} + + {id, {: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, + pending: MapSet.member?(pending, id) + }} + end) + end + + def frames(frames) do + Enum.map(frames, fn {file, line, name, code} -> + %{file: file, line: line, name: name, code: code} + end) + end + + def result(result, created_by \\ nil) do + created_by = build_principal(created_by) + + case result do + {:error, type, message, frames, retry, retryable} -> + %{ + type: "error", + createdBy: created_by, + error: %{type: type, message: message, frames: frames(frames)}, + retry: if(retry, do: execution_attempt(retry)), + retryable: retryable + } + + {:error, type, message, frames, retry} -> + %{ + type: "error", + createdBy: created_by, + error: %{type: type, message: message, frames: frames(frames)}, + retry: if(retry, do: execution_attempt(retry)), + retryable: nil + } + + {:value, value} -> + %{type: "value", createdBy: created_by, value: build_value(value)} + + {:abandoned, retry} -> + %{ + type: "abandoned", + createdBy: created_by, + retry: if(retry, do: execution_attempt(retry)) + } + + {:crashed, retry} -> + %{ + type: "crashed", + createdBy: created_by, + retry: if(retry, do: execution_attempt(retry)) + } + + :cancelled -> + %{type: "cancelled", createdBy: created_by} + + {:timeout, retry} -> + %{ + type: "timeout", + createdBy: created_by, + retry: if(retry, do: execution_attempt(retry)) + } + + {:suspended, successor} -> + %{ + type: "suspended", + createdBy: created_by, + successor: if(successor, do: execution_attempt(successor)) + } + + {:recurred, successor} -> + %{ + type: "recurred", + createdBy: created_by, + successor: if(successor, do: execution_attempt(successor)) + } + + {:deferred, execution, result} -> + %{ + type: "deferred", + createdBy: created_by, + execution: build_execution(execution), + result: result(result) + } + + {:cached, execution, result} -> + %{ + type: "cached", + createdBy: created_by, + execution: build_execution(execution), + result: result(result) + } + + {:spawned, execution, result} -> + %{ + type: "spawned", + createdBy: created_by, + execution: build_execution(execution), + result: result(result) + } + + nil -> + nil + end + end + + # Checkpoints are reported as a pair so an attempt can be read as a + # transition: `before` is what it was handed when it started, `after` what + # it ended up holding. They're equal for an attempt that didn't write. + def checkpoints(before, after_) do + %{before: values(before), after: values(after_)} + end + + # Streams belong to the step. Only those in a workspace the topic is + # showing are included — a re-run in another workspace opens its own. + def streams(streams, workspace_ids) do + streams + |> Enum.filter(fn {_index, stream} -> stream.workspace_id in workspace_ids end) + |> Map.new(fn {index, stream} -> {Integer.to_string(index), stream(stream)} end) + end + + def stream(stream) do + %{ + id: stream.id, + index: stream.index, + position: stream.position, + workspaceId: stream.workspace_id, + buffer: stream.buffer, + timeoutMs: stream.timeout_ms, + openedAt: stream.opened_at, + attempts: stream.attempts, + closedAt: stream.closed_at, + closedBy: stream.closed_by, + reason: stream.reason, + error: stream.error + } + end + + def stream_error(nil), do: nil + + def stream_error({type, message, frames}) do + %{type: type, message: message, frames: frames(frames)} + end + + defp execution_attempt({ext_id, _module, _target}) do + ext_id |> String.split(":") |> List.last() |> String.to_integer() + end +end diff --git a/server/lib/coflux/run_view/loader.ex b/server/lib/coflux/run_view/loader.ex new file mode 100644 index 00000000..3efefcc5 --- /dev/null +++ b/server/lib/coflux/run_view/loader.ex @@ -0,0 +1,42 @@ +defmodule Coflux.RunView.Loader do + @moduledoc """ + Subscribes a topic process to a run and builds the view from the snapshot. + """ + + alias Coflux.Orchestration + alias Coflux.RunView + + @doc """ + Returns `{:ok, view, run, parent, fetch}` with the view showing the + workspace the run started in plus the one being viewed, or + `{:error, :not_found}`. The view holds structure only; `fetch` loads the + detail `RunView.missing_details/3` asks for (see `RunView.Sync.load/4`). + """ + def load(project_id, run_id, workspace_id, pid) do + case Orchestration.subscribe_run(project_id, run_id, pid) do + {:error, :not_found} -> + {:error, :not_found} + + {:ok, run, parent, steps, _ref} -> + fetch = fn request -> + case Orchestration.get_run_details(project_id, run_id, request) do + {:ok, details} -> details + {:error, :not_found} -> %{executions: %{}, steps: %{}} + end + end + + run_workspace_id = + steps + |> Map.values() + |> Enum.reject(& &1.parent_id) + |> Enum.min_by(& &1.created_at) + |> Map.fetch!(:executions) + |> Map.values() + |> Enum.min_by(& &1.created_at) + |> Map.fetch!(:workspace_id) + + workspace_ids = Enum.uniq([run_workspace_id, workspace_id]) + {:ok, RunView.new(run, steps, workspace_ids), run, parent, fetch} + end + end +end diff --git a/server/lib/coflux/run_view/sync.ex b/server/lib/coflux/run_view/sync.ex new file mode 100644 index 00000000..e3c260fa --- /dev/null +++ b/server/lib/coflux/run_view/sync.ex @@ -0,0 +1,153 @@ +defmodule Coflux.RunView.Sync do + @moduledoc """ + Keeps a topic's `steps` in step with its view after a batch of + notifications has been applied. + + The topic's state carries the `view`, the set of `visible` steps, and a + `fetch` function that loads detail (`Orchestration.get_run_details`). + Steps that enter or leave the view are set or unset whole; steps that + stay are re-projected and diffed, so only what changed is sent. Detail + for anything about to be projected is fetched first if it isn't loaded. + + Which steps are visible is recomputed in full only when a visible step + gained an attempt (a re-run or retry can change which subtree is shown). + Anything else can only add: a new child link the view shows, or a first + attempt on a step that's already linked from a shown execution, brings + in the subtree it reaches. + """ + + alias Coflux.RunView + alias Coflux.RunView.Diff + alias Topical.Topic + + def steps(topic, effects) do + {added, removed, visible} = revisit(topic.state.view, topic.state.visible, effects.events) + + dirty = + effects.dirty + |> MapSet.intersection(visible) + |> MapSet.difference(added) + + topic = ensure(topic, MapSet.union(added, dirty)) + view = topic.state.view + + topic = + Enum.reduce(removed, topic, fn step, topic -> + Topic.unset(topic, [:steps], RunView.step_key(view, step)) + end) + + topic = + Enum.reduce(added, topic, fn step, topic -> + Topic.set(topic, [:steps, RunView.step_key(view, step)], RunView.project_step(view, step)) + end) + + topic = + Enum.reduce(dirty, topic, fn step, topic -> + key = RunView.step_key(view, step) + Diff.apply(topic, [:steps, key], topic.value.steps[key], RunView.project_step(view, step)) + end) + + %{topic | state: %{topic.state | visible: visible}} + end + + @doc "Replaces the whole `steps` map, for when the view's root or pins moved." + def reset(topic) do + visible = RunView.visible_steps(topic.state.view) + topic = ensure(topic, visible) + view = topic.state.view + topic = Diff.apply(topic, [:steps], topic.value.steps, RunView.project(view)) + %{topic | state: %{topic.state | visible: visible}} + end + + @doc """ + Keeps a structure topic's `steps` in step with its view: every step that + changed is re-projected and diffed. Steps only ever join (with their first + shown attempt), never leave, and no detail is involved. + """ + def structure(topic, effects) do + view = topic.state.view + + Enum.reduce(effects.dirty, topic, fn step, topic -> + if RunView.latest(view, step) do + key = RunView.step_key(view, step) + + Diff.apply( + topic, + [:steps, key], + topic.value.steps[key], + RunView.project_structure_step(view, step) + ) + else + topic + end + end) + end + + @doc "Loads any detail the given steps need before they can be projected." + def ensure(topic, steps, executions? \\ true) do + view = topic.state.view + request = RunView.missing_details(view, steps, executions?) + + if RunView.empty_request?(request) do + topic + else + view = RunView.put_details(view, topic.state.fetch.(request)) + %{topic | state: %{topic.state | view: view}} + end + end + + @doc "The view with detail loaded for the given steps, through `fetch`." + def load(view, fetch, steps, executions? \\ true) do + request = RunView.missing_details(view, steps, executions?) + + if RunView.empty_request?(request), + do: view, + else: RunView.put_details(view, fetch.(request)) + end + + defp revisit(view, visible, events) do + full? = + Enum.any?(events, fn + {:attempt, step} -> MapSet.member?(visible, step) + _ -> false + end) + + if full? do + now = RunView.visible_steps(view) + {MapSet.difference(now, visible), MapSet.difference(visible, now), now} + else + added = + Enum.reduce(events, MapSet.new(), fn + {:link, parent_id, step, group_id}, added -> + add_through(view, visible, added, parent_id, step, group_id) + + {:attempt, step}, added -> + view.parents + |> Map.get(step, []) + |> Enum.reduce(added, fn parent_id, added -> + group_id = view.child_index[parent_id][step] + add_through(view, visible, added, parent_id, step, group_id) + end) + + _event, added -> + added + end) + + {added, MapSet.new(), MapSet.union(visible, added)} + end + end + + # Adds what `step` reaches if the link to it from `parent_id` is one the + # view shows and its parent is shown. + defp add_through(view, visible, added, parent_id, step, group_id) do + seen = MapSet.union(visible, added) + parent_step = view.executions[parent_id].step + + if MapSet.member?(seen, parent_step) and not MapSet.member?(seen, step) and + RunView.link_visible?(view, parent_id, step, group_id) do + MapSet.union(added, RunView.reachable(view, step, seen)) + else + added + end + end +end diff --git a/server/lib/coflux/topics/run.ex b/server/lib/coflux/topics/run.ex index 1b2248e3..216a3ab2 100644 --- a/server/lib/coflux/topics/run.ex +++ b/server/lib/coflux/topics/run.ex @@ -1,837 +1,52 @@ defmodule Coflux.Topics.Run do - use Topical.Topic, route: ["workspaces", :workspace_id, "runs", :run_id] + @moduledoc """ + The skeleton of a run: the tree from its initial step, with the latest + attempt of each step expanded, other attempts summarised, and each group + collapsed to its first member plus a summary of the rest. Steps only + reachable through a superseded attempt aren't included. See + `Coflux.RunView` for the rule, and `Coflux.Topics.RunExecution` for + opening what's collapsed. + """ - alias Coflux.Orchestration + use Topical.Topic, route: ["workspaces", :workspace_id, "runs", :run_id] import Coflux.TopicUtils + alias Coflux.RunView + alias Coflux.RunView.{Loader, Sync} + def connect(params, context) do {:ok, Map.put(params, :project, context.project)} end def init(params) do project_id = Map.fetch!(params, :project) - external_run_id = Map.fetch!(params, :run_id) + run_id = Map.fetch!(params, :run_id) workspace_id = Map.fetch!(params, :workspace_id) - case Orchestration.subscribe_run( - project_id, - external_run_id, - self() - ) do + case Loader.load(project_id, run_id, workspace_id, self()) do {:error, :not_found} -> {:error, :not_found} - {:ok, run, parent, steps, _ref} -> - run_workspace_id = - steps - |> Map.values() - |> Enum.reject(& &1.parent_id) - |> Enum.min_by(& &1.created_at) - |> Map.fetch!(:executions) - |> Map.values() - |> Enum.min_by(& &1.created_at) - |> Map.fetch!(:workspace_id) - - workspace_ids = Enum.uniq([run_workspace_id, workspace_id]) - - {:ok, - Topic.new(build_run(run, parent, steps, workspace_ids), %{ - project: project_id, - external_run_id: external_run_id, - workspace_ids: workspace_ids - })} - end - end - - def handle_info({:topic, _ref, notifications}, topic) do - topic = Enum.reduce(notifications, topic, &process_notification(&2, &1)) - {:ok, topic} - end - - defp process_notification(topic, {:step, step_number, step, workspace_external_id}) do - if workspace_external_id in topic.state.workspace_ids do - Topic.set(topic, [:steps, "#{topic.state.external_run_id}:#{step_number}"], %{ - stepNumber: step_number, - module: step.module, - target: step.target, - parentId: step.parent_id, - cacheConfig: build_cache_config(step.cache_config), - cacheKey: build_key(step.cache_key), - memoKey: build_key(step.memo_key), - concurrency: build_concurrency(step), - group: build_group(step), - retries: build_retries(step.retries), - recurrent: step.recurrent, - timeout: step.timeout, - createdAt: step.created_at, - arguments: Enum.map(step.arguments, &build_value/1), - requires: step.requires, - streams: %{}, - executions: %{} - }) - else - topic - end - end - - defp process_notification( - topic, - {:execution, step_number, attempt, execution_external_id, workspace_external_id, - created_at, execute_after, dependencies, created_by, checkpoints, pending_dependencies} - ) do - if workspace_external_id in topic.state.workspace_ids do - Topic.set( - topic, - [ - :steps, - "#{topic.state.external_run_id}:#{step_number}", - :executions, - Integer.to_string(attempt) - ], - %{ - executionId: execution_external_id, - workspaceId: workspace_external_id, - createdAt: created_at, - createdBy: build_principal(created_by), - executeAfter: execute_after, - assignedAt: nil, - completedAt: nil, - completion: nil, - groups: %{}, - assets: %{}, - dependencies: - Map.new(dependencies, fn {dependency_id, dependency} -> - {dependency_id, - build_dependency( - dependency, - MapSet.member?(pending_dependencies, dependency_id) - )} - end), - children: [], - inputs: %{}, - result: nil, - metrics: %{}, - # Nothing has run yet, so what the execution will start from is also - # what it currently holds. - checkpoints: build_checkpoints(checkpoints, checkpoints) - } - ) - else - topic - end - end - - defp process_notification(topic, {:metric_defined, execution_external_id, key, definition}) do - update_execution(topic, execution_external_id, fn topic, base_path -> - Topic.set(topic, base_path ++ [:metrics, key], %{ - group: Map.get(definition, "group"), - groupUnits: Map.get(definition, "group_units"), - groupLower: Map.get(definition, "group_lower"), - groupUpper: Map.get(definition, "group_upper"), - scale: Map.get(definition, "scale"), - units: Map.get(definition, "units"), - progress: Map.get(definition, "progress", false), - lower: Map.get(definition, "lower"), - upper: Map.get(definition, "upper") - }) - end) - end - - defp process_notification( - topic, - {:group, execution_external_id, group_id, name, concurrency} - ) do - update_execution(topic, execution_external_id, fn topic, base_path -> - Topic.set(topic, base_path ++ [:groups, Integer.to_string(group_id)], %{ - name: name, - concurrency: concurrency - }) - end) - end - - defp process_notification(topic, {:asset, execution_external_id, external_asset_id, asset}) do - asset = build_asset(asset) - - update_execution( - topic, - execution_external_id, - fn topic, base_path -> - Topic.set( - topic, - base_path ++ [:assets, external_asset_id], - asset - ) - end - ) - end - - defp process_notification(topic, {:assigned, assigned}) do - Enum.reduce(assigned, topic, fn {execution_external_id, assigned_at}, topic -> - update_execution(topic, execution_external_id, fn topic, base_path -> - Topic.set(topic, base_path ++ [:assignedAt], assigned_at) - end) - 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, - {:pending_dependencies, execution_external_id, pending} - ) do - 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, - execution_external_id, - fn topic, base_path -> - Topic.merge( - topic, - base_path ++ [:dependencies, dependency_id], - dependency - ) - end - ) - end - - defp process_notification( - topic, - {:stream_dependency, execution_external_id, stream_id, module, target, pending} - ) do - 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) - end) - end - - defp process_notification(topic, {:child, parent_execution_external_id, child}) do - child = build_child(child, topic.state.external_run_id) - - update_execution(topic, parent_execution_external_id, fn topic, base_path -> - Topic.insert(topic, base_path ++ [:children], child) - end) - end - - defp process_notification( - topic, - {:result, execution_external_id, result, result_at, created_by} - ) do - result = build_result(result, created_by) - - update_execution(topic, execution_external_id, fn topic, base_path -> - topic - |> Topic.set(base_path ++ [:result], result) - |> Topic.set(base_path ++ [:resultAt], result_at) - end) - end - - defp process_notification( - topic, - {:completion, execution_external_id, kind, successor, completion_at} - ) do - topic - |> update_execution(execution_external_id, fn topic, base_path -> - topic - |> Topic.set(base_path ++ [:completedAt], completion_at) - |> Topic.set(base_path ++ [:completion], %{ - kind: Atom.to_string(kind), - 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 - # Only the "after" side moves — what the execution started from was fixed - # when it was created. The whole snapshot is replaced rather than merged: a - # reset drops a name entirely, so merging would leave it behind. - update_execution(topic, execution_external_id, fn topic, base_path -> - Topic.set(topic, base_path ++ [:checkpoints, :after], build_values(checkpoints)) - end) - end - - # An execution registered on one of the step's streams — opening it, or - # resuming it after a suspend (`continued`). Streams live under the step - # in the topic, with the attempts that have produced into each. - defp process_notification(topic, {:stream_registered, step_number, index, info}) do - step_key = "#{topic.state.external_run_id}:#{step_number}" - index_key = Integer.to_string(index) - path = [:steps, step_key, :streams, index_key] - - cond do - info.workspace_id not in topic.state.workspace_ids -> - topic - - not Map.has_key?(topic.value.steps, step_key) -> - topic - - true -> - case get_in(topic.value, path) do - nil -> - Topic.set(topic, path, %{ - id: info.id, - index: index, - position: info.position, - workspaceId: info.workspace_id, - buffer: info.buffer, - timeoutMs: info.timeout_ms, - openedAt: info.opened_at, - attempts: [info.attempt], - closedAt: nil, - closedBy: nil, - reason: nil, - error: nil - }) - - existing -> - attempts = - if info.attempt in existing.attempts, - do: existing.attempts, - else: existing.attempts ++ [info.attempt] - - topic - |> Topic.set(path ++ [:attempts], attempts) - |> Topic.set(path ++ [:buffer], info.buffer) - |> Topic.set(path ++ [:timeoutMs], info.timeout_ms) - end - end - end - - defp process_notification( - topic, - {:stream_closed, step_number, index, reason, error, attempt, closed_at} - ) do - step_key = "#{topic.state.external_run_id}:#{step_number}" - path = [:steps, step_key, :streams, Integer.to_string(index)] - - if get_in(topic.value, path) do - topic - |> Topic.set(path ++ [:closedAt], closed_at) - |> Topic.set(path ++ [:closedBy], attempt) - |> Topic.set(path ++ [:reason], reason) - |> Topic.set(path ++ [:error], error) - else - topic - end - end - - defp process_notification( - topic, - {:result_result, execution_external_id, result, _created_at, created_by} - ) do - result = build_result(result, created_by) - - update_execution(topic, execution_external_id, fn topic, base_path -> - Topic.set(topic, base_path ++ [:result, :result], result) - end) - end - - defp process_notification( - topic, - {:input_submitted, execution_external_id, input_external_id, title} - ) do - status = find_existing_input_status(topic, input_external_id) - - update_execution(topic, execution_external_id, fn topic, base_path -> - Topic.set(topic, base_path ++ [:inputs, input_external_id], %{ - title: title, - status: status - }) - end) - end - - defp process_notification( - topic, - {: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, - pending: pending - }) - end) - end - - defp process_notification( - topic, - {: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), - pending: pending - }) - end) - end - - defp process_notification( - topic, - {:input_response, input_external_id, response_type} - ) do - topic - |> update_submitted_input_status(input_external_id, response_type) - |> update_dependency_input_status(input_external_id, response_type) - end - - defp find_existing_input_status(topic, input_external_id) do - Enum.find_value(topic.value.steps, fn {_step_id, step} -> - Enum.find_value(step.executions, fn {_attempt, execution} -> - get_in(execution, [:inputs, input_external_id, :status]) - end) - end) - end - - defp update_submitted_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 -> - inputs = Map.get(execution, :inputs, %{}) + {:ok, view, run, parent, fetch} -> + visible = RunView.visible_steps(view) + view = Sync.load(view, fetch, visible) - if Map.has_key?(inputs, input_external_id) do - Topic.set( - topic, - [:steps, step_id, :executions, attempt, :inputs, input_external_id, :status], - response_type - ) - else - topic - end - end) - 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 -> - deps = Map.get(execution, :dependencies, %{}) - - if Map.has_key?(deps, input_external_id) and - Map.get(deps[input_external_id], :type) == "input" do - Topic.set( - topic, - [:steps, step_id, :executions, attempt, :dependencies, input_external_id, :status], - response_type - ) - else - topic - end - end) - end) - end - - defp build_run(run, parent, steps, workspace_ids) do - %{ - createdAt: run.created_at, - createdBy: build_principal(run.created_by), - requires: run.requires, - parent: if(parent, do: build_execution(parent)), - steps: - steps - |> Enum.filter(fn {_, step} -> - step.executions - |> Map.values() - |> Enum.any?(&(&1.workspace_id in workspace_ids)) - end) - |> Map.new(fn {step_id, step} -> - {"#{run.external_id}:#{step_id}", - %{ - stepNumber: step_id, - module: step.module, - target: step.target, - parentId: step.parent_id, - cacheConfig: build_cache_config(step.cache_config), - cacheKey: build_key(step.cache_key), - memoKey: build_key(step.memo_key), - concurrency: build_concurrency(step), - group: build_group(step), - retries: build_retries(step), - recurrent: step.recurrent == 1, - timeout: step.timeout, - createdAt: step.created_at, - arguments: Enum.map(step.arguments, &build_value/1), - requires: step.requires, - streams: build_streams(step.streams, workspace_ids), - executions: - step.executions - |> Enum.filter(fn {_, execution} -> - execution.workspace_id in workspace_ids - end) - |> Map.new(fn {attempt, execution} -> - {Integer.to_string(attempt), - %{ - executionId: execution.execution_id, - workspaceId: execution.workspace_id, - createdAt: execution.created_at, - createdBy: build_principal(execution.created_by), - executeAfter: execution.execute_after, - assignedAt: execution.assigned_at, - resultAt: execution.result_at, - completedAt: execution.completed_at, - completion: execution.completion, - groups: execution.groups, - assets: - Map.new(execution.assets, fn {external_asset_id, asset} -> - {external_asset_id, build_asset(asset)} - end), - 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), - metrics: - Map.new(execution.metric_definitions, fn {key, def_data} -> - {key, - %{ - group: def_data.group, - groupUnits: def_data.group_units, - groupLower: def_data.group_lower, - groupUpper: def_data.group_upper, - scale: def_data.scale, - units: def_data.units, - progress: def_data.progress, - lower: def_data.lower, - upper: def_data.upper - }} - end), - checkpoints: - build_checkpoints( - execution.checkpoints.before, - execution.checkpoints.after - ) - }} - end) - }} - end) - } - end - - # `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), - pending: MapSet.member?(pending, id) - }} - - {id, {:input, title, 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), - pending: MapSet.member?(pending, id) - }} - - {id, {:stream, stream_id, module, target}} -> - {id, - %{ - type: "stream", - streamId: stream_id, - module: module, - target: target, - pending: MapSet.member?(pending, id) - }} - end) - end - - defp build_dependency(execution, pending) do - %{ - type: "result", - execution: build_execution(execution), - pending: pending - } - end - - defp build_frames(frames) do - Enum.map(frames, fn {file, line, name, code} -> - %{ - file: file, - line: line, - name: name, - code: code - } - end) - end - - defp build_result(result, created_by \\ nil) do - created_by = build_principal(created_by) - - case result do - {:error, type, message, frames, retry, retryable} -> - %{ - type: "error", - createdBy: created_by, - error: %{ - type: type, - message: message, - frames: build_frames(frames) - }, - retry: if(retry, do: execution_attempt(retry)), - retryable: retryable - } - - {:error, type, message, frames, retry} -> - %{ - type: "error", - createdBy: created_by, - error: %{ - type: type, - message: message, - frames: build_frames(frames) - }, - retry: if(retry, do: execution_attempt(retry)), - retryable: nil - } - - {:value, value} -> - %{type: "value", createdBy: created_by, value: build_value(value)} - - {:abandoned, retry} -> - %{ - type: "abandoned", - createdBy: created_by, - retry: if(retry, do: execution_attempt(retry)) - } - - {:crashed, retry} -> - %{ - type: "crashed", - createdBy: created_by, - retry: if(retry, do: execution_attempt(retry)) - } - - :cancelled -> - %{type: "cancelled", createdBy: created_by} - - {:timeout, retry} -> - %{ - type: "timeout", - createdBy: created_by, - retry: if(retry, do: execution_attempt(retry)) + value = %{ + createdAt: run.created_at, + createdBy: build_principal(run.created_by), + requires: run.requires, + parent: if(parent, do: build_execution(parent)), + steps: RunView.project(view) } - {:suspended, successor} -> - %{ - type: "suspended", - createdBy: created_by, - successor: if(successor, do: execution_attempt(successor)) - } - - {:recurred, successor} -> - %{ - type: "recurred", - createdBy: created_by, - successor: if(successor, do: execution_attempt(successor)) - } - - {:deferred, execution, result} -> - %{ - type: "deferred", - createdBy: created_by, - execution: build_execution(execution), - result: build_result(result) - } - - {:cached, execution, result} -> - %{ - type: "cached", - createdBy: created_by, - execution: build_execution(execution), - result: build_result(result) - } - - {:spawned, execution, result} -> - %{ - type: "spawned", - createdBy: created_by, - execution: build_execution(execution), - result: build_result(result) - } - - nil -> - nil + {:ok, Topic.new(value, %{view: view, visible: visible, fetch: fetch})} end end - # Checkpoints are reported as a pair so an attempt can be read as a - # transition: `before` is what it was handed when it started, `after` what it - # ended up holding. They're equal for an attempt that didn't write. - defp build_checkpoints(before, after_) do - %{before: build_values(before), after: build_values(after_)} - end - - defp build_values(values) do - Map.new(values, fn {name, value} -> {name, build_value(value)} end) - end - - # Streams belong to the step. Only those in a workspace the topic is - # showing are included — a re-run in another workspace opens its own. - defp build_streams(streams, workspace_ids) do - streams - |> Enum.filter(fn {_index, stream} -> stream.workspace_id in workspace_ids end) - |> Map.new(fn {index, stream} -> - {Integer.to_string(index), - %{ - id: stream.id, - index: stream.index, - position: stream.position, - workspaceId: stream.workspace_id, - buffer: stream.buffer, - timeoutMs: stream.timeout_ms, - openedAt: stream.opened_at, - attempts: stream.attempts, - closedAt: stream.closed_at, - closedBy: stream.closed_by, - reason: if(stream.reason, do: Atom.to_string(stream.reason)), - error: build_stream_error(stream.error) - }} - end) - end - - defp build_stream_error(nil), do: nil - - defp build_stream_error({type, message, frames}) do - %{type: type, message: message, frames: build_frames(frames)} - end - - defp execution_attempt({ext_id, _module, _target}) do - ext_id |> String.split(":") |> List.last() |> String.to_integer() - end - - defp build_child({step_number, attempt, group_id}, external_run_id) do - %{stepId: "#{external_run_id}:#{step_number}", attempt: attempt, groupId: group_id} - end - - # From notification map (retries opt from worker) - defp build_retries(nil), do: nil - - defp build_retries(%{limit: limit, backoff_min: backoff_min, backoff_max: backoff_max}) do - %{limit: limit, backoffMin: backoff_min, backoffMax: backoff_max} - end - - # From Step struct - defp build_retries(%{retry_limit: 0}), do: nil - - defp build_retries(%{ - retry_limit: retry_limit, - retry_backoff_min: retry_backoff_min, - retry_backoff_max: retry_backoff_max - }) do - %{ - limit: if(retry_limit == -1, do: nil, else: retry_limit), - backoffMin: retry_backoff_min, - backoffMax: retry_backoff_max - } - end - - defp build_cache_config(cache_config) do - if cache_config do - %{ - params: cache_config.params, - maxAge: cache_config.max_age, - namespace: cache_config.namespace, - version: cache_config.version - } - end - end - - defp build_concurrency(step) do - if step.concurrency_key do - %{limit: step.concurrency_limit, key: build_key(step.concurrency_key)} - end - end - - # The group limit the step counts against, if any. The key is readable - # as it stands ("/"), so no hex prefix. - defp build_group(step) do - if step.group_key do - %{limit: step.group_limit, key: step.group_key} - end - end - - defp build_key(key, length \\ 10) do - if key do - key - |> Base.encode16(case: :lower) - |> String.slice(0, length) - end - end - - defp update_execution(topic, execution_external_id, fun) 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 - fun.(topic, [:steps, step_id, :executions, attempt]) - else - topic - end - end) - end) + def handle_info({:topic, _ref, notifications}, topic) do + {view, effects} = RunView.apply_all(topic.state.view, notifications) + topic = %{topic | state: %{topic.state | view: view}} + {:ok, Sync.steps(topic, effects)} end end diff --git a/server/lib/coflux/topics/run_execution.ex b/server/lib/coflux/topics/run_execution.ex new file mode 100644 index 00000000..4c137e3b --- /dev/null +++ b/server/lib/coflux/topics/run_execution.ex @@ -0,0 +1,104 @@ +defmodule Coflux.Topics.RunExecution do + @moduledoc """ + The part of a run's view that changes when an execution is pinned: an old + attempt, or a group member other than the first. The value is the tree + from the topmost step at which the pinned path leaves the default view, + with the path pinned below it, plus the hop from the default view into + that root. Studio merges it over the run topic by step. An execution + that's in the default view already gives an empty value. + """ + + use Topical.Topic, + route: ["workspaces", :workspace_id, "runs", :run_id, "executions", :execution_id] + + alias Coflux.RunView + alias Coflux.RunView.{Loader, Sync} + + def connect(params, context) do + {:ok, Map.put(params, :project, context.project)} + end + + def init(params) do + project_id = Map.fetch!(params, :project) + run_id = Map.fetch!(params, :run_id) + workspace_id = Map.fetch!(params, :workspace_id) + execution_id = Map.fetch!(params, :execution_id) + + with {:ok, view, _run, _parent, fetch} <- + Loader.load(project_id, run_id, workspace_id, self()), + %{} = selection <- RunView.selection(view, execution_id) do + view = RunView.with_selection(view, selection) + visible = RunView.visible_steps(view) + view = Sync.load(view, fetch, visible) + + value = %{ + root: root_id(view), + parent: hop(selection.parent), + steps: RunView.project(view) + } + + state = %{ + view: view, + visible: visible, + selection: selection, + execution_id: execution_id, + fetch: fetch + } + + {:ok, Topic.new(value, state)} + else + nil -> {:error, :not_found} + {:error, :not_found} -> {:error, :not_found} + end + end + + def handle_info({:topic, _ref, notifications}, topic) do + {view, effects} = RunView.apply_all(topic.state.view, notifications) + topic = %{topic | state: %{topic.state | view: view}} + previous = topic.state.selection + + # A change on the pinned path (a new attempt above the root, a new + # parent link) can move where the view diverges from the default. Then + # the pins and root are rederived and the value replaced. + if RunView.touches?(view, effects.events, previous.path) do + selection = RunView.selection(view, topic.state.execution_id) + + if selection.root == previous.root and selection.pins == previous.pins and + selection.parent == previous.parent do + {:ok, Sync.steps(topic, effects)} + else + view = RunView.with_selection(view, selection) + topic = %{topic | state: %{topic.state | view: view, selection: selection}} + + topic = + topic + |> Sync.reset() + |> Topic.set([:root], root_id(view)) + |> Topic.set([:parent], hop(selection.parent)) + + {:ok, topic} + end + else + {:ok, Sync.steps(topic, effects)} + end + end + + defp hop(nil), do: nil + + defp hop(%{execution_id: execution_id, group_id: group_id}) do + %{executionId: execution_id, groupId: group_id} + end + + defp root_id(view) do + case RunView.root_step(view) do + nil -> + nil + + step -> + case RunView.expanded_execution(view, step) do + nil -> nil + execution -> execution.id + end + end + end +end diff --git a/server/lib/coflux/topics/run_group.ex b/server/lib/coflux/topics/run_group.ex new file mode 100644 index 00000000..9b5cfc9b --- /dev/null +++ b/server/lib/coflux/topics/run_group.ex @@ -0,0 +1,130 @@ +defmodule Coflux.Topics.RunGroup do + @moduledoc """ + The members of one group, for the group popover: every child submitted + into the group by that execution, with its branch status. The run topic + only carries the group's first member and a summary of the rest. + + Members are keyed by step under `members`, with submission order kept in + `order`, so a member's status or attempt changes by key and the order + only ever grows by appending — no update names a position in a list. + """ + + use Topical.Topic, + route: [ + "workspaces", + :workspace_id, + "runs", + :run_id, + "executions", + :execution_id, + "groups", + :group_id + ] + + alias Coflux.RunView + alias Coflux.RunView.{Loader, Sync} + + def connect(params, context) do + {:ok, Map.put(params, :project, context.project)} + end + + def init(params) do + project_id = Map.fetch!(params, :project) + run_id = Map.fetch!(params, :run_id) + workspace_id = Map.fetch!(params, :workspace_id) + execution_id = Map.fetch!(params, :execution_id) + + with {group_id, ""} <- Integer.parse(Map.fetch!(params, :group_id)), + {:ok, view, _run, _parent, fetch} <- + Loader.load(project_id, run_id, workspace_id, self()), + %{groups: %{^group_id => group}} <- Map.get(view.executions, execution_id) do + # Members are listed with their arguments, so those are loaded for + # every member step (and nothing else). + member_steps = + view.children + |> Map.get(execution_id, []) + |> Enum.filter(&(&1.group_id == group_id)) + |> Enum.map(& &1.step) + + view = Sync.load(view, fetch, member_steps, false) + members = RunView.group_members(view, execution_id, group_id) + + value = %{ + name: group.name, + concurrency: group.concurrency, + members: Map.new(members, &{&1.stepId, &1}), + order: Enum.map(members, & &1.stepId) + } + + state = %{ + view: view, + execution_id: execution_id, + group_id: group_id, + fetch: fetch + } + + {:ok, Topic.new(value, state)} + else + _ -> {:error, :not_found} + end + end + + def handle_info({:topic, _ref, notifications}, topic) do + {view, effects} = RunView.apply_all(topic.state.view, notifications) + topic = %{topic | state: %{topic.state | view: view}} + %{execution_id: execution_id, group_id: group_id} = topic.state + + topic = + Enum.reduce(effects.events, topic, fn + {:link, ^execution_id, step, ^group_id}, topic -> + topic = Sync.ensure(topic, [step], false) + view = topic.state.view + link = Enum.find(view.children[execution_id], &(&1.step == step)) + member = RunView.group_member(view, link) + + topic + |> Topic.set([:members, member.stepId], member) + |> Topic.insert([:order], member.stepId) + + _event, topic -> + topic + end) + + topic = + Enum.reduce(effects.branches, topic, fn step, topic -> + update_member(topic, view, step, :status, RunView.member_status(view, step)) + end) + + topic = + Enum.reduce(effects.events, topic, fn + {:attempt, step}, topic -> + update_member(topic, view, step, :attempt, RunView.latest(view, step)) + + _event, topic -> + topic + end) + + group = view.executions[execution_id].groups[group_id] + + topic = + topic + |> maybe_set([:name], group.name) + |> maybe_set([:concurrency], group.concurrency) + + {:ok, topic} + end + + defp update_member(topic, view, step, field, value) do + key = RunView.step_key(view, step) + + case Map.fetch(topic.value.members, key) do + {:ok, %{^field => ^value}} -> topic + {:ok, _member} -> Topic.set(topic, [:members, key, field], value) + :error -> topic + end + end + + defp maybe_set(topic, [key] = path, value) do + if Map.get(topic.value, key) == value, do: topic, else: Topic.set(topic, path, value) + end +end diff --git a/server/lib/coflux/topics/run_steps.ex b/server/lib/coflux/topics/run_steps.ex new file mode 100644 index 00000000..a3f9ffd7 --- /dev/null +++ b/server/lib/coflux/topics/run_steps.ex @@ -0,0 +1,38 @@ +defmodule Coflux.Topics.RunSteps do + @moduledoc """ + The structure of a whole run: every step, each attempt summarised with + its complete list of children. Nothing is collapsed and no detail is + carried, so it stays small even for a large fan-out, and the pages that + show the whole run (timeline, logs) can traverse it while the run topic + keeps the graph's collapsed view. + """ + + use Topical.Topic, route: ["workspaces", :workspace_id, "runs", :run_id, "steps"] + + alias Coflux.RunView + alias Coflux.RunView.{Loader, Sync} + + def connect(params, context) do + {:ok, Map.put(params, :project, context.project)} + end + + def init(params) do + project_id = Map.fetch!(params, :project) + run_id = Map.fetch!(params, :run_id) + workspace_id = Map.fetch!(params, :workspace_id) + + case Loader.load(project_id, run_id, workspace_id, self()) do + {:error, :not_found} -> + {:error, :not_found} + + {:ok, view, _run, _parent, _fetch} -> + {:ok, Topic.new(%{steps: RunView.project_structure(view)}, %{view: view})} + end + end + + def handle_info({:topic, _ref, notifications}, topic) do + {view, effects} = RunView.apply_all(topic.state.view, notifications) + topic = %{topic | state: %{topic.state | view: view}} + {:ok, Sync.structure(topic, effects)} + end +end diff --git a/server/test/coflux/run_view_test.exs b/server/test/coflux/run_view_test.exs new file mode 100644 index 00000000..0973add1 --- /dev/null +++ b/server/test/coflux/run_view_test.exs @@ -0,0 +1,717 @@ +defmodule Coflux.RunViewTest do + use ExUnit.Case, async: true + + alias Coflux.RunView + alias Coflux.RunView.{Diff, Sync} + alias Topical.Topic + + @run %{external_id: "R1", created_at: 0, created_by: nil, requires: %{}} + @ws "W1" + + # --------------------------------------------------------------------------- + # Snapshot builders (the shape `build_run_data` produces) + + defp id(number, attempt), do: "R1:#{number}:#{attempt}" + + # A snapshot step. With `detail: false` it carries structure only, as + # `subscribe_run` returns it: no arguments or streams, and executions + # without results, dependencies and the rest. + defp step(number, opts \\ []) do + detail? = Keyword.get(opts, :detail, true) + + base = %{ + module: "m", + target: "t#{number}", + type: "task", + parent_id: Keyword.get(opts, :parent), + cache_config: nil, + cache_key: nil, + memo_key: nil, + concurrency_key: nil, + concurrency_limit: 0, + group_key: nil, + group_limit: 0, + retry_limit: 0, + retry_backoff_min: 0, + retry_backoff_max: 0, + recurrent: 0, + timeout: 0, + created_at: number * 10, + requires: %{}, + executions: + Map.new(Keyword.get(opts, :executions, []), fn {attempt, execution_opts} -> + {attempt, execution(number, attempt, [{:detail, detail?} | execution_opts])} + end) + } + + if detail?, do: Map.merge(base, %{arguments: [], streams: %{}}), else: base + end + + defp execution(number, attempt, opts) do + base = %{ + execution_id: id(number, attempt), + workspace_id: Keyword.get(opts, :ws, @ws), + created_at: Keyword.get(opts, :created_at, number * 10 + attempt), + created_by: nil, + execute_after: nil, + assigned_at: Keyword.get(opts, :assigned_at), + completed_at: Keyword.get(opts, :completed_at), + completion: + case Keyword.get(opts, :completion) do + nil -> nil + kind -> %{kind: kind, successor: nil} + end, + groups: Keyword.get(opts, :groups, %{}), + children: Keyword.get(opts, :children, []) + } + + if Keyword.get(opts, :detail, true), + do: Map.merge(base, detail(Keyword.get(opts, :dependencies, %{}))), + else: base + end + + # Detail as `get_run_details` returns it. + defp detail(dependencies \\ %{}, result \\ nil) do + %{ + assets: %{}, + dependencies: dependencies, + pending_dependencies: MapSet.new(), + inputs: %{}, + result: result, + result_at: nil, + result_created_by: nil, + metric_definitions: %{}, + checkpoints: %{before: %{}, after: %{}} + } + end + + # 1 (running) + # ├─ 2 (completed) + # └─ group 1 "batch": 3 (running) ─ 6 (assigning) + # 4 (errored) + # 5 (assigning) + defp fixture(detail? \\ true) do + steps = %{ + 1 => + step(1, + detail: detail?, + executions: [ + {1, + [ + assigned_at: 1, + groups: %{1 => %{name: "batch", concurrency: 0}}, + children: [{2, 1, nil}, {3, 1, 1}, {4, 1, 1}, {5, 1, 1}] + ]} + ] + ), + 2 => + step(2, + detail: detail?, + parent: id(1, 1), + executions: [{1, [assigned_at: 1, completed_at: 2, completion: "succeeded"]}] + ), + 3 => + step(3, + detail: detail?, + parent: id(1, 1), + executions: [{1, [assigned_at: 1, children: [{6, 1, nil}]]}] + ), + 4 => + step(4, + detail: detail?, + parent: id(1, 1), + executions: [{1, [assigned_at: 1, completed_at: 2, completion: "errored"]}] + ), + 5 => step(5, detail: detail?, parent: id(1, 1), executions: [{1, []}]), + 6 => step(6, detail: detail?, parent: id(3, 1), executions: [{1, []}]) + } + + RunView.new(@run, steps, [@ws]) + end + + # --------------------------------------------------------------------------- + # Notification builders + + defp n_step(number, parent) do + {:step, number, + %{ + module: "m", + target: "t#{number}", + type: "task", + parent_id: parent, + cache_config: nil, + cache_key: nil, + memo_key: nil, + concurrency_key: nil, + concurrency_limit: 0, + group_key: nil, + group_limit: 0, + retries: nil, + recurrent: false, + timeout: 0, + created_at: number * 10, + arguments: [], + requires: %{} + }, @ws} + end + + defp n_execution(number, attempt, opts \\ []) do + {:execution, number, attempt, id(number, attempt), Keyword.get(opts, :ws, @ws), + Keyword.get(opts, :created_at, number * 10 + attempt), nil, %{}, nil, %{}, MapSet.new()} + end + + defp n_child(parent_id, number, attempt, group_id) do + {:child, parent_id, {number, attempt, group_id}} + end + + defp n_assigned(execution_id), do: {:assigned, %{execution_id => 100}} + defp n_completion(execution_id, kind), do: {:completion, execution_id, kind, nil, 200} + + # --------------------------------------------------------------------------- + # A topic kept in sync, and oracles + + defp topic(view, fetch \\ fn request -> flunk("unexpected fetch: #{inspect(request)}") end) do + Topic.new(%{steps: RunView.project(view)}, %{ + view: view, + visible: RunView.visible_steps(view), + fetch: fetch + }) + end + + # A fetch that answers with arguments naming the step and a value result, + # and records what it was asked for. + defp recording_fetch(log) do + fn request -> + Agent.update(log, &[request | &1]) + + %{ + executions: + Map.new(request.executions, fn {number, attempt} -> + {id(number, attempt), detail(%{}, {:value, {:raw, "r#{number}", []}})} + end), + steps: + Map.new(request.steps, fn number -> + {number, %{arguments: [{:raw, "arg#{number}", []}], streams: %{}}} + end) + } + end + end + + defp sync(topic, notifications) do + {view, effects} = RunView.apply_all(topic.state.view, notifications) + topic = %{topic | state: %{topic.state | view: view}} + Sync.steps(topic, effects) + end + + defp assert_in_sync(topic) do + assert topic.value.steps == RunView.project(topic.state.view) + assert topic.state.visible == RunView.visible_steps(topic.state.view) + assert_branches(topic.state.view) + end + + # A structure topic kept in sync alongside + defp structure_topic(view) do + Topic.new(%{steps: RunView.project_structure(view)}, %{view: view}) + end + + defp sync_structure(topic, notifications) do + {view, effects} = RunView.apply_all(topic.state.view, notifications) + topic = %{topic | state: %{topic.state | view: view}} + Sync.structure(topic, effects) + end + + # Studio's `getBranchStatus`: the latest attempt's status combined with + # every child's branch, recursively. + defp assert_branches(view) do + for step <- Map.keys(view.steps), RunView.latest(view, step) do + statuses = oracle_statuses(view, step, MapSet.new()) + + expected = + Enum.find( + [:running, :assigning, :errored, :aborted, :suspended], + :completed, + &(&1 in statuses) + ) + + assert RunView.branch_status(view, step) == expected, "branch status of step #{step}" + end + end + + defp oracle_statuses(view, step, seen) do + case RunView.latest(view, step) do + nil -> + [] + + attempt -> + execution = view.executions[view.attempts[step][attempt]] + seen = MapSet.put(seen, step) + + children = + view.child_index + |> Map.get(execution.id, %{}) + |> Map.keys() + |> Enum.reject(&MapSet.member?(seen, &1)) + + [own_status(execution) | Enum.flat_map(children, &oracle_statuses(view, &1, seen))] + end + end + + defp own_status(%{completion: nil, assigned_at: nil}), do: :assigning + defp own_status(%{completion: nil}), do: :running + + defp own_status(%{completion: %{kind: kind}}) do + case kind do + "succeeded" -> :completed + "errored" -> :errored + "cancelled" -> :aborted + "suspended" -> :suspended + end + end + + # --------------------------------------------------------------------------- + + describe "the default view" do + test "shows the latest tree with each group collapsed to its first member" do + view = fixture() + assert RunView.visible_steps(view) == MapSet.new([1, 2, 3, 6]) + + steps = RunView.project(view) + assert steps |> Map.keys() |> Enum.sort() == ["R1:1", "R1:2", "R1:3", "R1:6"] + + root = steps["R1:1"] + assert Map.keys(root.attempts) == ["1"] + assert Map.keys(root.executions) == ["1"] + execution = root.executions["1"] + assert Enum.map(execution.children, & &1.stepId) == ["R1:2", "R1:3"] + + assert execution.groups["1"] == %{ + name: "batch", + concurrency: 0, + members: %{ + total: 3, + byStatus: %{ + assigning: 1, + running: 1, + errored: 1, + completed: 0, + aborted: 0, + suspended: 0 + } + } + } + + assert steps["R1:3"].attempts["1"] == %{ + executionId: "R1:3:1", + workspaceId: @ws, + createdAt: 31, + executeAfter: nil, + assignedAt: 1, + completedAt: nil, + completion: nil + } + end + + test "branch status rolls up through latest attempts and every group member" do + view = fixture() + assert_branches(view) + assert RunView.branch_status(view, 1) == :running + assert RunView.branch_status(view, 3) == :running + + {view, _} = + RunView.apply_all(view, [ + n_completion(id(1, 1), :succeeded), + n_completion(id(3, 1), :succeeded) + ]) + + assert RunView.branch_status(view, 3) == :assigning + assert RunView.branch_status(view, 1) == :assigning + + {view, effects} = + RunView.apply_all(view, [ + n_assigned(id(6, 1)), + n_completion(id(6, 1), :succeeded), + n_assigned(id(5, 1)), + n_completion(id(5, 1), :succeeded) + ]) + + assert RunView.branch_status(view, 1) == :errored + # The root's group summary moved, so its entry is dirty + assert MapSet.member?(effects.dirty, 1) + assert MapSet.equal?(effects.branches, MapSet.new([1, 3, 5, 6])) + assert_branches(view) + + members = RunView.project(view)["R1:1"].executions["1"].groups["1"].members + + assert members.byStatus == %{ + completed: 2, + errored: 1, + assigning: 0, + running: 0, + aborted: 0, + suspended: 0 + } + end + + test "lists a group's members in order with their branch status" do + members = RunView.group_members(fixture(), id(1, 1), 1) + + assert Enum.map(members, &{&1.stepId, &1.status}) == [ + {"R1:3", "running"}, + {"R1:4", "errored"}, + {"R1:5", "assigning"} + ] + end + end + + describe "keeping a topic in sync" do + test "a new attempt is expanded and what only the old one reached is dropped" do + topic = sync(topic(fixture()), [n_execution(3, 2)]) + view = topic.state.view + + assert RunView.visible_steps(view) == MapSet.new([1, 2, 3]) + assert topic.value.steps["R1:3"].attempts |> Map.keys() |> Enum.sort() == ["1", "2"] + assert Map.keys(topic.value.steps["R1:3"].executions) == ["2"] + refute Map.has_key?(topic.value.steps, "R1:6") + assert RunView.branch_status(view, 3) == :assigning + assert topic.value.steps["R1:1"].executions["1"].groups["1"].members.byStatus.assigning == 2 + assert_in_sync(topic) + end + + test "a new child appears under an expanded execution and is counted under a collapsed group" do + topic = + sync(topic(fixture()), [ + n_step(7, id(2, 1)), + n_execution(7, 1), + n_child(id(2, 1), 7, 1, nil) + ]) + + assert Map.has_key?(topic.value.steps, "R1:7") + assert Enum.map(topic.value.steps["R1:2"].executions["1"].children, & &1.stepId) == ["R1:7"] + assert_in_sync(topic) + + topic = sync(topic, [n_step(8, id(1, 1)), n_execution(8, 1), n_child(id(1, 1), 8, 1, 1)]) + refute Map.has_key?(topic.value.steps, "R1:8") + assert topic.value.steps["R1:1"].executions["1"].groups["1"].members.total == 4 + assert_in_sync(topic) + + # A memo hit linking that member under a shown execution brings it in + topic = sync(topic, [n_child(id(2, 1), 8, 1, nil)]) + assert Map.has_key?(topic.value.steps, "R1:8") + assert_in_sync(topic) + end + + test "an attempt in another workspace is ignored until one lands in a shown workspace" do + topic = + sync(topic(fixture()), [ + n_step(7, id(2, 1)), + n_execution(7, 1, ws: "W2"), + n_child(id(2, 1), 7, 1, nil) + ]) + + refute Map.has_key?(topic.value.steps, "R1:7") + assert Map.has_key?(topic.state.view.steps, 7) + assert topic.value.steps["R1:1"].executions["1"].groups["1"].members.total == 3 + assert_in_sync(topic) + + topic = sync(topic, [n_execution(7, 2)]) + assert Map.keys(topic.value.steps["R1:7"].attempts) == ["2"] + assert_in_sync(topic) + end + + test "matches a fresh projection through a sequence of batches" do + batches = [ + [n_assigned(id(5, 1))], + [n_step(7, id(3, 1)), n_execution(7, 1), n_child(id(3, 1), 7, 1, nil)], + [n_assigned(id(7, 1)), n_completion(id(7, 1), :errored)], + [n_execution(7, 2)], + [n_assigned(id(7, 2)), n_completion(id(7, 2), :succeeded)], + [n_execution(3, 2)], + [n_completion(id(3, 2), :succeeded)], + [n_step(8, id(3, 2)), n_execution(8, 1), n_child(id(3, 2), 8, 1, nil)], + [n_child(id(3, 2), 6, 1, nil)], + [n_completion(id(1, 1), :succeeded)], + [n_execution(1, 2)], + [ + n_step(9, id(1, 2)), + n_execution(9, 1), + n_child(id(1, 2), 9, 1, nil), + n_child(id(1, 2), 3, 2, nil) + ] + ] + + Enum.reduce(batches, topic(fixture()), fn batch, topic -> + topic = sync(topic, batch) + assert_in_sync(topic) + topic + end) + end + + test "input responses reach submissions and dependencies" do + topic = + sync(topic(fixture()), [ + {:input_submitted, id(3, 1), "I1", "Pick"}, + {:input_dependency, id(2, 1), "I1", "Pick", nil, true} + ]) + + assert topic.value.steps["R1:3"].executions["1"].inputs == %{ + "I1" => %{title: "Pick", status: nil} + } + + assert topic.value.steps["R1:2"].executions["1"].dependencies["I1"].pending == true + + topic = sync(topic, [{:input_response, "I1", "value"}]) + assert topic.value.steps["R1:3"].executions["1"].inputs["I1"].status == "value" + assert topic.value.steps["R1:2"].executions["1"].dependencies["I1"].status == "value" + assert_in_sync(topic) + + # A later submission of the same input starts with its status + topic = sync(topic, [{:input_submitted, id(6, 1), "I1", "Pick"}]) + assert topic.value.steps["R1:6"].executions["1"].inputs["I1"].status == "value" + assert_in_sync(topic) + end + end + + describe "the structure" do + test "carries every step with all attempts and complete children, and nothing else" do + view = fixture(false) + steps = RunView.project_structure(view) + + assert steps |> Map.keys() |> Enum.sort() == [ + "R1:1", + "R1:2", + "R1:3", + "R1:4", + "R1:5", + "R1:6" + ] + + root = steps["R1:1"].attempts["1"] + + assert Enum.map(root.children, &{&1.stepId, &1.groupId}) == [ + {"R1:2", nil}, + {"R1:3", 1}, + {"R1:4", 1}, + {"R1:5", 1} + ] + + assert root.completion == nil + refute Map.has_key?(steps["R1:1"], :executions) + refute Map.has_key?(steps["R1:1"], :arguments) + end + + test "stays in sync as steps, attempts and links arrive" do + batches = [ + [n_execution(3, 2)], + [n_completion(id(3, 2), :succeeded)], + [n_step(7, id(3, 2)), n_execution(7, 1), n_child(id(3, 2), 7, 1, nil)], + [n_step(8, id(1, 1)), n_execution(8, 1, ws: "W2"), n_child(id(1, 1), 8, 1, 1)], + [n_execution(8, 2)] + ] + + topic = + Enum.reduce(batches, structure_topic(fixture(false)), fn batch, topic -> + topic = sync_structure(topic, batch) + assert topic.value.steps == RunView.project_structure(topic.state.view) + topic + end) + + assert topic.value.steps["R1:3"].attempts |> Map.keys() |> Enum.sort() == ["1", "2"] + assert Enum.map(topic.value.steps["R1:3"].attempts["2"].children, & &1.stepId) == ["R1:7"] + # Step 8's first attempt was in another workspace; it joined with its second + assert Map.keys(topic.value.steps["R1:8"].attempts) == ["2"] + + assert Enum.map(topic.value.steps["R1:1"].attempts["1"].children, & &1.stepId) == [ + "R1:2", + "R1:3", + "R1:4", + "R1:5", + "R1:8" + ] + end + end + + describe "loading detail on demand" do + test "a structure-only snapshot reports what the visible steps need" do + view = fixture(false) + assert RunView.visible_steps(view) == MapSet.new([1, 2, 3, 6]) + + request = RunView.missing_details(view, RunView.visible_steps(view)) + assert Enum.sort(request.steps) == [1, 2, 3, 6] + assert Enum.sort(request.executions) == [{1, 1}, {2, 1}, {3, 1}, {6, 1}] + + # Only arguments, when asked for members + assert RunView.missing_details(view, [4, 5], false) == %{executions: [], steps: [5, 4]} + + # Status is structure, so it's known without any detail + assert_branches(view) + assert RunView.project(view)["R1:1"].executions["1"].groups["1"].members.total == 3 + end + + test "a topic loads detail for what it shows, and again for what comes into view" do + {:ok, log} = Agent.start_link(fn -> [] end) + fetch = recording_fetch(log) + view = fixture(false) + view = Sync.load(view, fetch, RunView.visible_steps(view)) + topic = topic(view, fetch) + + assert [%{steps: steps, executions: executions}] = Agent.get(log, & &1) + assert Enum.sort(steps) == [1, 2, 3, 6] + assert length(executions) == 4 + assert topic.value.steps["R1:2"].arguments == [%{type: "raw", data: "arg2", references: []}] + assert topic.value.steps["R1:2"].executions["1"].result.value.data == "r2" + assert_in_sync(topic) + + # A status change on a loaded step needs nothing more + topic = sync(topic, [n_completion(id(3, 1), :succeeded)]) + assert length(Agent.get(log, & &1)) == 1 + assert_in_sync(topic) + + # A memo hit brings the unloaded member 4 into view under the leaf + topic = sync(topic, [n_child(id(2, 1), 4, 1, nil)]) + assert [%{steps: [4], executions: [{4, 1}]} | _] = Agent.get(log, & &1) + assert topic.value.steps["R1:4"].arguments == [%{type: "raw", data: "arg4", references: []}] + assert_in_sync(topic) + + # A new attempt arrives with its detail, so nothing is fetched for it + topic = sync(topic, [n_execution(2, 2)]) + assert length(Agent.get(log, & &1)) == 2 + assert Map.keys(topic.value.steps["R1:2"].executions) == ["2"] + assert_in_sync(topic) + end + + test "a pinned view re-rooting loads the attempt it uncovers" do + {:ok, log} = Agent.start_link(fn -> [] end) + fetch = recording_fetch(log) + view = fixture(false) + selection = RunView.selection(view, id(4, 1)) + view = RunView.with_selection(view, selection) + view = Sync.load(view, fetch, RunView.visible_steps(view)) + topic = topic(view, fetch) + assert Map.keys(topic.value.steps) == ["R1:4"] + + # The root is re-run: the view moves up to the root's old attempt, + # whose detail wasn't loaded + {view, _effects} = RunView.apply_all(topic.state.view, [n_execution(1, 2)]) + selection = RunView.selection(view, id(4, 1)) + assert selection.root == {:step, 1} + view = RunView.with_selection(view, selection) + topic = %{topic | state: %{topic.state | view: view}} + topic = Sync.reset(topic) + + assert Map.keys(topic.value.steps) |> Enum.sort() == ["R1:1", "R1:2", "R1:4"] + assert [%{steps: steps, executions: executions} | _] = Agent.get(log, & &1) + assert Enum.sort(steps) == [1, 2] + assert Enum.sort(executions) == [{1, 1}, {2, 1}] + assert topic.value.steps["R1:1"].executions["1"].result.value.data == "r1" + assert_in_sync(topic) + end + end + + describe "selection" do + test "an execution in the default view needs nothing" do + view = fixture() + assert %{root: :none, parent: nil} = RunView.selection(view, id(2, 1)) + assert %{root: :none, parent: nil} = RunView.selection(view, id(6, 1)) + assert RunView.selection(view, "R1:9:1") == nil + end + + test "a group member other than the first roots the view at it" do + view = fixture() + selection = RunView.selection(view, id(4, 1)) + assert selection.root == {:step, 4} + assert selection.parent == %{execution_id: id(1, 1), group_id: 1} + assert selection.pins.groups == %{{id(1, 1), 1} => 4} + assert selection.path == MapSet.new([1, 4]) + + view = RunView.with_selection(view, selection) + assert RunView.visible_steps(view) == MapSet.new([4]) + end + + test "a superseded attempt roots the view at its step and shows what it reached" do + {view, _} = RunView.apply_all(fixture(), [n_execution(3, 2)]) + selection = RunView.selection(view, id(3, 1)) + assert selection.root == {:step, 3} + assert selection.parent == %{execution_id: id(1, 1), group_id: 1} + assert selection.pins.steps == %{3 => 1, 1 => 1} + + view = RunView.with_selection(view, selection) + assert RunView.visible_steps(view) == MapSet.new([3, 6]) + assert Map.keys(RunView.project(view)["R1:3"].executions) == ["1"] + end + + test "a re-run above a pinned execution moves the root up" do + view = fixture() + selection = RunView.selection(view, id(4, 1)) + + {view, effects} = RunView.apply_all(view, [n_execution(1, 2)]) + assert RunView.touches?(view, effects.events, selection.path) + + selection = RunView.selection(view, id(4, 1)) + assert selection.root == {:step, 1} + assert selection.parent == nil + assert selection.pins.steps == %{1 => 1, 4 => 1} + + view = RunView.with_selection(view, selection) + assert RunView.visible_steps(view) == MapSet.new([1, 2, 4]) + assert Map.keys(RunView.project(view)["R1:1"].executions) == ["1"] + end + + test "the path follows the most recently created parent, and a memo hit counts once per parent" do + {view, _} = + RunView.apply_all(fixture(), [ + n_execution(2, 2, created_at: 500), + n_child(id(2, 2), 6, 1, nil) + ]) + + assert view.parents[6] == [id(3, 1), id(2, 2)] + assert view.child_counts[id(2, 2)] == %{assigning: 1} + + selection = RunView.selection(view, id(6, 1)) + assert selection.pins.steps == %{6 => 1, 2 => 2, 1 => 1} + assert selection.root == :none + end + + test "a pinned view stays in sync as the run moves on" do + view = fixture() + selection = RunView.selection(view, id(4, 1)) + view = RunView.with_selection(view, selection) + topic = topic(view) + + topic = + sync(topic, [ + n_step(7, id(4, 1)), + n_execution(7, 1), + n_child(id(4, 1), 7, 1, nil), + n_assigned(id(4, 1)) + ]) + + assert Map.keys(topic.value.steps) |> Enum.sort() == ["R1:4", "R1:7"] + assert_in_sync(topic) + + # A retry of the pinned step doesn't move the pin + topic = sync(topic, [n_execution(4, 2)]) + assert Map.keys(topic.value.steps["R1:4"].executions) == ["1"] + assert Map.has_key?(topic.value.steps, "R1:7") + assert_in_sync(topic) + end + end + + describe "diff" do + test "sends only what changed" do + old = %{x: 1, y: %{z: 1, w: 2}, l: [1], gone: true} + new = %{x: 1, y: %{z: 2}, l: [1, 2], n: 3} + topic = Diff.apply(Topic.new(%{steps: %{"a" => old}}), [:steps, "a"], old, new) + + assert topic.value.steps["a"] == new + + assert Enum.sort(topic.updates) == + Enum.sort([ + {:unset, [:steps, "a"], :gone}, + {:unset, [:steps, "a", :y], :w}, + {:set, [:steps, "a", :y, :z], 2}, + {:set, [:steps, "a", :l], [1, 2]}, + {:set, [:steps, "a", :n], 3} + ]) + end + end +end diff --git a/tests/conftest.py b/tests/conftest.py index 72fa0c57..981aab6f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -6,7 +6,13 @@ import pytest from support import cli -from support.helpers import ADAPTER_SCRIPT, managed_worker, poll_result +from support.helpers import ( + ADAPTER_SCRIPT, + get_topic, + managed_worker, + poll_result, + workspace_id, +) from support.server import ManagedServer @@ -48,9 +54,46 @@ def result(self, run_id, timeout=10): ) def inspect(self, run_id): - """Get the full run topic snapshot.""" + """Get the run topic snapshot: the latest tree, with groups collapsed.""" return cli.runs_inspect(run_id, host=self.host, workspace=self.workspace) + def inspect_execution(self, run_id, execution_id): + """Get the execution topic snapshot: what pinning the execution opens.""" + return get_topic( + self.host, + "workspaces", + workspace_id(self.host, self.workspace), + "runs", + run_id, + "executions", + execution_id, + ) + + def inspect_steps(self, run_id): + """Get the steps topic snapshot: every step's structure, nothing collapsed.""" + return get_topic( + self.host, + "workspaces", + workspace_id(self.host, self.workspace), + "runs", + run_id, + "steps", + ) + + def inspect_group(self, run_id, execution_id, group_id): + """Get the group topic snapshot: every member of the group.""" + return get_topic( + self.host, + "workspaces", + workspace_id(self.host, self.workspace), + "runs", + run_id, + "executions", + execution_id, + "groups", + group_id, + ) + def queue(self): """Get the queue topic snapshot for the workspace.""" return cli.queue(host=self.host, workspace=self.workspace) diff --git a/tests/support/helpers.py b/tests/support/helpers.py index 0eff6ca1..efb3d972 100644 --- a/tests/support/helpers.py +++ b/tests/support/helpers.py @@ -35,6 +35,19 @@ def poll_result( ) +def get_topic(host, *route): + """Capture a topic snapshot over the REST endpoint.""" + url = f"http://{host}/topics/" + "/".join(str(part) for part in route) + with urllib.request.urlopen(urllib.request.Request(url), timeout=5) as resp: + return json.load(resp) + + +def workspace_id(host, name="default"): + """The external id of the workspace called ``name``.""" + workspaces = get_topic(host, "workspaces") + return next(k for k, v in workspaces.items() if v["name"] == name) + + def api_post(port, project_id, path, token=None, body=None): """POST to a server management API endpoint, returning the decoded reply.""" if token is None: diff --git a/tests/test_checkpoints.py b/tests/test_checkpoints.py index 9e658b1d..8f53c1e3 100644 --- a/tests/test_checkpoints.py +++ b/tests/test_checkpoints.py @@ -513,20 +513,25 @@ def test_exposed_in_run_topic(worker): ex1.conn.complete(ex1.execution_id, value="done") ctx.result(run_id) - executions = next(iter(ctx.inspect(run_id)["steps"].values()))["executions"] + # The run topic expands the latest attempt; the first is opened by + # the execution topic. + step_id, step = next(iter(ctx.inspect(run_id)["steps"].items())) + second = step["executions"]["2"] + opened = ctx.inspect_execution(run_id, ex0.execution_id) + first = opened["steps"][step_id]["executions"]["1"] # Each attempt is reported as a transition: what it was handed when it # started, and what it ended up holding. The first attempt started from # nothing; the second inherited both names and reset one of them, so the # reset shows up as an absence on the "after" side only. - assert executions["1"]["checkpoints"]["before"] == {} - assert executions["1"]["checkpoints"]["after"]["cursor"]["data"] == 5 - assert executions["1"]["checkpoints"]["after"]["batch"]["data"] == "a" - - assert executions["2"]["checkpoints"]["before"]["cursor"]["data"] == 5 - assert executions["2"]["checkpoints"]["before"]["batch"]["data"] == "a" - assert executions["2"]["checkpoints"]["after"]["cursor"]["data"] == 5 - assert "batch" not in executions["2"]["checkpoints"]["after"] + assert first["checkpoints"]["before"] == {} + assert first["checkpoints"]["after"]["cursor"]["data"] == 5 + assert first["checkpoints"]["after"]["batch"]["data"] == "a" + + assert second["checkpoints"]["before"]["cursor"]["data"] == 5 + assert second["checkpoints"]["before"]["batch"]["data"] == "a" + assert second["checkpoints"]["after"]["cursor"]["data"] == 5 + assert "batch" not in second["checkpoints"]["after"] def test_run_topic_reports_untouched_state(worker): diff --git a/tests/test_logging.py b/tests/test_logging.py index f5e930ce..67c3ec9c 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -75,12 +75,19 @@ def test_execution_groups(worker): ex0.conn.complete(ex0.execution_id, value="grouped") - # Verify groups show in inspect + # The run topic collapses the group to its first member plus a + # summary; the group topic lists every member. data = ctx.inspect(run_id) child_steps = { sid: s for sid, s in data["steps"].items() if s.get("target") == "item" } - assert len(child_steps) == 2 + assert len(child_steps) == 1 + root_step = next(s for s in data["steps"].values() if s["parentId"] is None) + group = next(iter(root_step["executions"].values()))["groups"]["1"] + assert group["name"] == "batch" + assert group["members"]["total"] == 2 + members = ctx.inspect_group(run_id, ex0.execution_id, 1) + assert len(members["order"]) == 2 assert ctx.result(run_id)["value"]["data"] == "grouped" diff --git a/tests/test_run_topic.py b/tests/test_run_topic.py new file mode 100644 index 00000000..d0d773ee --- /dev/null +++ b/tests/test_run_topic.py @@ -0,0 +1,203 @@ +"""Tests for the run topics. + +The run topic is a skeleton: the tree from the initial step with the latest +attempt of each step expanded, other attempts summarised, and each group +collapsed to its first member plus a summary of the rest. The execution +topic opens what's collapsed for one pinned execution, and the group topic +lists a group's members. +""" + +import time + +from support.manifest import task, workflow +from support.protocol import json_args, register_group_notification + + +def wait_for(fn, timeout=5, interval=0.1): + deadline = time.time() + timeout + while True: + value = fn() + if value: + return value + if time.time() > deadline: + raise TimeoutError("condition not met") + time.sleep(interval) + + +def step_number(execution_id): + return int(execution_id.split(":")[1]) + + +def counts(**by_status): + return { + "total": sum(by_status.values()), + "byStatus": { + **{ + s: 0 + for s in ( + "assigning", + "running", + "completed", + "errored", + "aborted", + "suspended", + ) + }, + **by_status, + }, + } + + +def test_run_topics(worker): + targets = [ + workflow("test", "main"), + task("test", "item", parameters=["x"]), + task("test", "sub"), + task("test", "leaf"), + ] + + with worker(targets, concurrency=8) as ctx: + run_id = ctx.submit("test", "main")["runId"] + root = ctx.executor.next_execute() + + # Steps are numbered in submission order: items are 2, 3, 4 and the + # ungrouped leaf is 5. + root.conn.send( + register_group_notification(root.execution_id, group_id=0, name="batch") + ) + for x in (1, 2, 3): + root.conn.submit_task( + root.execution_id, "test", "item", json_args(x), group_id=0 + ) + root.conn.submit_task(root.execution_id, "test", "leaf", json_args()) + + by_step = {} + for _ in range(4): + ex = ctx.executor.next_execute(timeout=5) + by_step[step_number(ex.execution_id)] = ex + assert set(by_step) == {2, 3, 4, 5} + + # The second item has a child of its own (step 6). + by_step[3].conn.submit_task(by_step[3].execution_id, "test", "sub", json_args()) + sub = ctx.executor.next_execute(timeout=5) + assert step_number(sub.execution_id) == 6 + + sub.conn.complete(sub.execution_id, value="sub") + by_step[3].conn.complete(by_step[3].execution_id, value="b") + by_step[4].conn.complete(by_step[4].execution_id, value="c") + by_step[5].conn.complete(by_step[5].execution_id, value="leaf") + by_step[2].conn.fail(by_step[2].execution_id, "ValueError", "boom") + root.conn.complete(root.execution_id, value="done") + assert ctx.result(run_id)["value"]["data"] == "done" + + def run_topic(): + return ctx.inspect(run_id) + + def execution_topic(execution_id): + return ctx.inspect_execution(run_id, execution_id) + + def group_topic(): + return ctx.inspect_group(run_id, root.execution_id, 0) + + def key(step): + return f"{run_id}:{step}" + + # --- The skeleton: root, the group's first member, and the leaf. --- + # Children complete over their own connections, so the root's result + # can land before every child's completion has been processed. + def settled(steps): + members = steps[key(1)]["executions"]["1"]["groups"]["0"]["members"] + return members == counts(completed=2, errored=1) + + steps = wait_for( + lambda: (lambda s: s if settled(s) else None)(run_topic()["steps"]) + ) + assert set(steps) == {key(1), key(2), key(5)} + + root_execution = steps[key(1)]["executions"]["1"] + assert [c["stepId"] for c in root_execution["children"]] == [key(2), key(5)] + assert root_execution["groups"]["0"]["name"] == "batch" + assert root_execution["groups"]["0"]["members"] == counts( + completed=2, errored=1 + ) + assert set(steps[key(2)]["attempts"]) == {"1"} + assert steps[key(2)]["attempts"]["1"]["completion"]["kind"] == "errored" + + # --- A member other than the first is opened by the execution topic. --- + opened = execution_topic(by_step[3].execution_id) + assert opened["root"] == by_step[3].execution_id + assert opened["parent"] == {"executionId": root.execution_id, "groupId": 0} + assert set(opened["steps"]) == {key(3), key(6)} + + # --- Something already in the skeleton needs nothing. --- + assert execution_topic(by_step[5].execution_id) == { + "root": None, + "parent": None, + "steps": {}, + } + + # --- The steps topic has every step, with complete children. --- + structure = ctx.inspect_steps(run_id)["steps"] + assert set(structure) == {key(n) for n in range(1, 7)} + root_attempt = structure[key(1)]["attempts"]["1"] + assert [(c["stepId"], c["groupId"]) for c in root_attempt["children"]] == [ + (key(2), 0), + (key(3), 0), + (key(4), 0), + (key(5), None), + ] + assert structure[key(2)]["attempts"]["1"]["completion"]["kind"] == "errored" + assert "executions" not in structure[key(1)] + + # --- The group topic lists every member with its branch status. --- + group = group_topic() + assert group["name"] == "batch" + assert group["order"] == [key(2), key(3), key(4)] + assert [(s, group["members"][s]["status"]) for s in group["order"]] == [ + (key(2), "errored"), + (key(3), "completed"), + (key(4), "completed"), + ] + + # --- A re-run expands the new attempt and summarises the old one. --- + assert ctx.rerun(key(2))["attempt"] == 2 + retry = ctx.executor.next_execute(timeout=5) + assert retry.execution_id == f"{run_id}:2:2" + retry.conn.complete(retry.execution_id, value="a") + + def retried(steps): + attempt = steps[key(2)]["attempts"].get("2") + return attempt and attempt["completion"] + + steps = wait_for( + lambda: (lambda s: s if retried(s) else None)(run_topic()["steps"]) + ) + assert set(steps[key(2)]["attempts"]) == {"1", "2"} + assert set(steps[key(2)]["executions"]) == {"2"} + assert steps[key(2)]["attempts"]["2"]["completion"]["kind"] == "succeeded" + assert steps[key(1)]["executions"]["1"]["groups"]["0"]["members"] == counts( + completed=3 + ) + + # The steps topic gained the attempt too + structure = ctx.inspect_steps(run_id)["steps"] + assert set(structure[key(2)]["attempts"]) == {"1", "2"} + + # The old attempt can still be opened, on its own. + opened = execution_topic(f"{run_id}:2:1") + assert opened["root"] == f"{run_id}:2:1" + assert opened["parent"] == {"executionId": root.execution_id, "groupId": 0} + assert set(opened["steps"]) == {key(2)} + assert set(opened["steps"][key(2)]["executions"]) == {"1"} + assert set(opened["steps"][key(2)]["attempts"]) == {"1", "2"} + + # And the group's member status and attempt follow the latest attempt. + group = wait_for( + lambda: ( + lambda g: g if g["members"][key(2)]["status"] == "completed" else None + )(group_topic()) + ) + assert [group["members"][s]["status"] for s in group["order"]] == [ + "completed" + ] * 3 + assert group["members"][key(2)]["attempt"] == 2 diff --git a/tests/test_streams.py b/tests/test_streams.py index 0b5e4ede..21ffb0ab 100644 --- a/tests/test_streams.py +++ b/tests/test_streams.py @@ -2245,9 +2245,12 @@ def test_stream_dependency_reported_for_each_consumer_attempt(worker): cons_resp = ctx.submit("test", "consumer") cons_ex = ctx.executor.next_execute() - # Open the topic before anything subscribes, so every edge below has + # Open the topics before anything subscribes, so every edge below has # to arrive as an update rather than being read from the snapshot. + # The run topic will show the latest attempt; the execution topic + # for the first attempt shows that one once it's been superseded. ctx.inspect(cons_resp["runId"]) + ctx.inspect_execution(cons_resp["runId"], cons_ex.execution_id) cons_ex.conn.stream_subscribe( cons_ex.execution_id, subscription_id=1, stream_id=stream["id"] @@ -2279,7 +2282,8 @@ def test_stream_dependency_reported_for_each_consumer_attempt(worker): "pending": False, } } - _, step = next(iter(ctx.inspect(cons_resp["runId"])["steps"].items())) - assert sorted(step["executions"]) == ["1", "2"] - for attempt, execution in step["executions"].items(): - assert execution["dependencies"] == expected, f"attempt {attempt}" + step_id, step = next(iter(ctx.inspect(cons_resp["runId"])["steps"].items())) + assert sorted(step["attempts"]) == ["1", "2"] + assert step["executions"]["2"]["dependencies"] == expected + opened = ctx.inspect_execution(cons_resp["runId"], cons_ex.execution_id) + assert opened["steps"][step_id]["executions"]["1"]["dependencies"] == expected