From bfd5c8572b867de5fa32fab76e192f1dd2c2d134 Mon Sep 17 00:00:00 2001 From: Kasun Vithanage Date: Fri, 23 Oct 2026 07:21:09 +0530 Subject: [PATCH 1/3] fix(core): store string step errors as maps instead of leaving the run running A step returning {:error, "declined"} handed the raw string to the failed/compensated write. The error column is a map, so the write was rejected: without compensations the run stayed :running until stale-lock recovery; with them, the executor crashed. handle_step_failure/4 now normalizes and sanitizes the error, as the parallel-child path already did. --- durable/lib/durable/executor.ex | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/durable/lib/durable/executor.ex b/durable/lib/durable/executor.ex index e9446d1..60d6128 100644 --- a/durable/lib/durable/executor.ex +++ b/durable/lib/durable/executor.ex @@ -1854,6 +1854,11 @@ defmodule Durable.Executor do # Handles step failure by running compensations if needed defp handle_step_failure(execution, error, workflow_def, config) do + # `error` is a map column: a step's `{:error, "declined"}` must become a + # map here, or the failed/compensated write is rejected and the run stays + # :running. + error = error |> normalize_error() |> sanitize_for_json() + # Get the compensation stack from completed steps compensation_stack = build_compensation_stack(execution.id, workflow_def, config) From 609c7e4d11d2d09cd9d1f2d466354be94fae97a8 Mon Sep 17 00:00:00 2001 From: Kasun Vithanage Date: Fri, 23 Oct 2026 07:21:10 +0530 Subject: [PATCH 2/3] feat(core): add Durable.Testing helpers for testing workflows run_workflow/3 and drain/2 run a workflow inline in the test process, including parallel branches and child workflows, until nothing is left to do. Sleeps end immediately and retry backoff is skipped. assert_status, assert_step, assert_waiting_for_input and assert_waiting_for_event give failure messages that show the run's current step and error. Adds guides/testing.md to HexDocs, the README and the site. --- durable/README.md | 1 + durable/guides/testing.md | 198 +++++++++++++ durable/lib/durable/executor/backoff.ex | 12 +- durable/lib/durable/executor/step_runner.ex | 2 +- durable/lib/durable/testing.ex | 293 ++++++++++++++++++++ durable/mix.exs | 1 + durable/test/durable/testing_test.exs | 134 +++++++++ site/src/docs.ts | 1 + 8 files changed, 639 insertions(+), 3 deletions(-) create mode 100644 durable/guides/testing.md create mode 100644 durable/lib/durable/testing.ex create mode 100644 durable/test/durable/testing_test.exs diff --git a/durable/README.md b/durable/README.md index 719920b..620ca25 100644 --- a/durable/README.md +++ b/durable/README.md @@ -619,6 +619,7 @@ option list and design notes. - [Compensations](guides/compensations.md) - Saga pattern - [Waiting](guides/waiting.md) - Sleep, events, human input - [Orchestration](guides/orchestration.md) - Parent/child workflow composition +- [Testing](guides/testing.md) - Run and assert on workflows in ExUnit ## License diff --git a/durable/guides/testing.md b/durable/guides/testing.md new file mode 100644 index 0000000..418ce49 --- /dev/null +++ b/durable/guides/testing.md @@ -0,0 +1,198 @@ +# Testing + +Test workflows by running them inline in the test process. `Durable.Testing` +runs a workflow, and everything it spawns, until nothing is left to do, then +gives you the run to assert on. + +## Setup + +Turn queue processing off in `config/test.exs`. Queue workers would otherwise +race your tests for the same runs: + +```elixir +config :durable, disable_queue_processing: true +``` + +Durable still starts from your supervision tree as usual. Then import the +helpers, for example in your `DataCase`: + +```elixir +defmodule MyApp.DataCase do + use ExUnit.CaseTemplate + + using do + quote do + import Durable.Testing + # ... + end + end +end +``` + +Runs execute in the test process, so they share the test's Ecto sandbox and +everything is rolled back when the test ends. + +## Running a Workflow + +`run_workflow/3` starts a workflow and drains it. It returns the run as a +`%Durable.Storage.Schemas.WorkflowExecution{}`: + +```elixir +test "fulfils an order" do + run = run_workflow(MyApp.OrderWorkflow, %{"order_id" => 42}) + + assert_status(run, :completed) + assert %{"order_id" => 42, "shipped" => true} = run.context +end +``` + +It takes the same options as `Durable.start/3`, such as `workflow:` to pick +a workflow in a module that defines several. + +Context and input are stored as JSON, so read them with string keys. + +`assert_status/2` checks `:completed`, `:failed`, `:waiting`, `:cancelled`, +`:compensated` or any other status. When it fails, it prints the run's +current step and error: + +``` +Expected workflow run 0f3c… to be :completed, got :failed +current step: "charge" +error: %{"message" => "card declined", "type" => "error"} +``` + +## Steps + +`assert_step/3` checks how a step's latest attempt ended (`:completed` by +default) and returns its `%StepExecution{}`: + +```elixir +assert_step(run, :reserve) +assert_step(run, :charge, :failed) + +assert %{"invoice_id" => _} = assert_step(run, :invoice).output +``` + +Steps inside `parallel` and `branch` blocks are matched by their own name, so +`assert_step/3` also tells you which branch ran: + +```elixir +run = run_workflow(MyApp.FulfilmentWorkflow, %{"type" => "digital"}) + +assert_step(run, :send_download_link) +``` + +`steps/2` lists every step execution, oldest first, with one row per attempt. + +## Parallel Blocks + +`drain/2` runs every branch of a parallel block, then the steps after it. +Unless the block has an `into:` callback, the branch results are stored under +`__results__`: + +```elixir +run = run_workflow(MyApp.OnboardingWorkflow, %{}) + +assert %{"__results__" => %{"send_email" => ["ok", _]}} = run.context +``` + +## Waits + +A run that waits for input or an event stops at `:waiting`. Resume it with the +regular API, then drain it again: + +```elixir +test "a manager approves the expense" do + run = run_workflow(MyApp.ExpenseWorkflow, %{"amount" => 500}) + + assert_status(run, :waiting) + assert_waiting_for_input(run, "manager_approval") + + :ok = Durable.provide_input(run.id, "manager_approval", %{"approved" => true}) + + run + |> drain() + |> assert_status(:completed) +end +``` + +Events work the same way, with `assert_waiting_for_event/2` and +`Durable.send_event/3`. + +`sleep/1` and `schedule_at/1` don't hold a test up: drain ends them right +away, so a run that sleeps for a day finishes in milliseconds. + +## Retries + +Retries skip their backoff during a drain. `attempt` on the step's latest row +says how many attempts it took: + +```elixir +test "retries a flaky gateway" do + expect(MyApp.GatewayMock, :charge, 2, fn _ -> {:error, :timeout} end) + expect(MyApp.GatewayMock, :charge, fn _ -> {:ok, "ch_1"} end) + + run = run_workflow(MyApp.PaymentWorkflow, %{"amount" => 100}) + + assert_status(run, :completed) + assert assert_step(run, :charge).attempt == 3 +end +``` + +Steps run in the test process, so Mox expectations apply without +`allow/3`. + +## Compensations + +A failure that's rolled back ends as `:compensated`, or as +`:compensation_failed` if a compensation failed too. Compensation steps are +recorded as `compensate_`: + +```elixir +run = run_workflow(MyApp.BookTripWorkflow, %{}) + +assert_status(run, :compensated) +assert_step(run, :book_hotel, :failed) +assert_step(run, :compensate_cancel_flight) +``` + +## Child Workflows + +Drain also runs the child workflows that a run starts, with `call_workflow/3` +or `start_workflow/3`, and resumes the parent once a child is done. Use +`Durable.list_children/2` to inspect them. + +## Scheduled Workflows + +A scheduled workflow is an ordinary workflow, so test it with +`run_workflow/3`. To test the schedule itself, create it and trigger it: + +```elixir +{:ok, _} = Durable.schedule(MyApp.ReportWorkflow, "0 9 * * *", name: "daily_report") +{:ok, id} = Durable.trigger_schedule("daily_report") + +id +|> drain() +|> assert_status(:completed) +``` + +## Limits + +- Wait timeouts (`timeout:` on `wait_for_input/2` and the other waits) don't + fire during a drain, because the timeout worker only runs with the queue + enabled. +- A workflow's `timeout:` uses the real clock, so drain doesn't skip ahead to + it. +- `drain/2` raises if the Durable instance has queue processing enabled. To + test with real queue workers, start Durable with the queue on and poll + `Durable.get_execution/2` until the run finishes. + +## Multiple Instances + +Every helper takes a `durable:` option naming the Durable instance, which +defaults to `Durable`: + +```elixir +run_workflow(MyApp.OrderWorkflow, %{}, durable: MyApp.Durable) +assert_status(run, :completed, durable: MyApp.Durable) +``` diff --git a/durable/lib/durable/executor/backoff.ex b/durable/lib/durable/executor/backoff.ex index cf1dcde..9b41d35 100644 --- a/durable/lib/durable/executor/backoff.ex +++ b/durable/lib/durable/executor/backoff.ex @@ -119,8 +119,16 @@ defmodule Durable.Executor.Backoff do """ @spec sleep(strategy(), pos_integer(), opts()) :: :ok def sleep(strategy, attempt, opts \\ %{}) do - delay = calculate_with_jitter(strategy, attempt, opts) - Process.sleep(delay) + strategy + |> calculate_with_jitter(attempt, opts) + |> pause() + end + + @doc false + # `Durable.Testing.drain/2` sets `:durable_skip_backoff` so retries don't slow tests. + @spec pause(non_neg_integer()) :: :ok + def pause(ms) do + unless Process.get(:durable_skip_backoff), do: Process.sleep(ms) :ok end end diff --git a/durable/lib/durable/executor/step_runner.ex b/durable/lib/durable/executor/step_runner.ex index 0632936..4edd20d 100644 --- a/durable/lib/durable/executor/step_runner.ex +++ b/durable/lib/durable/executor/step_runner.ex @@ -255,7 +255,7 @@ defmodule Durable.Executor.StepRunner do |> Map.put(:error, error_summary(error)) ) - Process.sleep(backoff_ms) + Backoff.pause(backoff_ms) execute_with_retry(step, data, step_exec.workflow_id, attempt + 1, max_attempts, config) else diff --git a/durable/lib/durable/testing.ex b/durable/lib/durable/testing.ex new file mode 100644 index 0000000..7c19d12 --- /dev/null +++ b/durable/lib/durable/testing.ex @@ -0,0 +1,293 @@ +defmodule Durable.Testing do + @moduledoc """ + Helpers for testing workflows. + + Runs execute inline in the test process, with no queue or pollers, so they + share the test's Ecto sandbox. `drain/2` keeps running a workflow and every + run it spawns until nothing is left to do. Sleeps end immediately and + retries skip their backoff. + + ## Setup + + Turn queue processing off in `config/test.exs`, so pollers don't race the + test for the same runs: + + config :durable, disable_queue_processing: true + + Then import the helpers where you test workflows, for example in your + `DataCase`: + + import Durable.Testing + + ## Example + + test "charges the order" do + run = run_workflow(OrderWorkflow, %{"order_id" => 42}) + + assert_status(run, :completed) + assert_step(run, :charge) + assert %{"charged" => true} = run.context + end + + A waiting run is resumed with the regular API, then drained: + + run = run_workflow(ApprovalWorkflow, %{}) + assert_waiting_for_input(run, "manager_approval") + + :ok = Durable.provide_input(run.id, "manager_approval", %{"approved" => true}) + + run + |> drain() + |> assert_status(:completed) + + Every function takes a `:durable` option that names the Durable instance + (default `Durable`). Runs can be passed as a `%WorkflowExecution{}`, a map + with an `:id`, or the id itself. + """ + + import Ecto.Query + import ExUnit.Assertions, only: [flunk: 1] + + alias Durable.Config + alias Durable.Executor + alias Durable.Queue.Adapter + alias Durable.Repo + alias Durable.Storage.Schemas.{PendingEvent, PendingInput, StepExecution, WorkflowExecution} + + @max_rounds 100 + + @doc """ + Starts the workflow in `module` with `input` and drains it. Returns the run + as a `%WorkflowExecution{}`, finished or waiting. + + Takes the same options as `Durable.start/3`. + """ + @spec run_workflow(module(), map(), keyword()) :: WorkflowExecution.t() + def run_workflow(module, input \\ %{}, opts \\ []) do + case Durable.start(module, input, opts) do + {:ok, id} -> + drain(id, opts) + + {:error, reason} -> + raise ArgumentError, "can't start #{inspect(module)}: #{inspect(reason)}" + end + end + + @doc """ + Runs every pending execution in the run's tree (the run, its parallel + branches and its child workflows) until none is left, then returns the + reloaded run. + + Sleeps end immediately and retries skip their backoff. Use it after + `Durable.provide_input/3` or `Durable.send_event/3`. + """ + @spec drain(WorkflowExecution.t() | map() | String.t(), keyword()) :: WorkflowExecution.t() + def drain(run, opts \\ []) do + config = config(opts) + id = run_id(run) + + if config.queue_enabled do + raise ArgumentError, """ + Durable.Testing.drain/2 needs queue processing off, or queue workers race \ + the test for the same runs. Add `config :durable, disable_queue_processing: true` \ + to config/test.exs, or start Durable with `queue_enabled: false`. + """ + end + + previous = Process.put(:durable_skip_backoff, true) + + try do + drain_rounds(config, id, @max_rounds) + after + Process.put(:durable_skip_backoff, previous) + end + + Repo.get(config, WorkflowExecution, id) + end + + defp drain_rounds(_config, id, 0), + do: flunk("Workflow run #{id} still had work after #{@max_rounds} drain rounds") + + defp drain_rounds(config, id, rounds) do + ids = tree_ids(config, [id]) + wake_sleepers(config, ids) + + pending = + Repo.all( + config, + from(w in WorkflowExecution, + where: w.id in ^ids and w.status == :pending, + order_by: w.inserted_at, + select: w.id + ) + ) + + if pending != [] do + Enum.each(pending, &Executor.execute_workflow(&1, config)) + drain_rounds(config, id, rounds - 1) + end + end + + defp tree_ids(config, ids) do + children = + Repo.all( + config, + from(w in WorkflowExecution, + where: w.parent_workflow_id in ^ids and w.id not in ^ids, + select: w.id + ) + ) + + if children == [], do: ids, else: tree_ids(config, ids ++ children) + end + + # Back-dates the tree's sleeps so the adapter's wake sweep, which takes the + # oldest first, wakes exactly these rows. + defp wake_sleepers(config, ids) do + sleeping = + from(w in WorkflowExecution, + where: w.id in ^ids and w.status == :waiting and not is_nil(w.scheduled_at) + ) + + case Repo.update_all(config, sleeping, set: [scheduled_at: ~U[2000-01-01 00:00:00.000000Z]]) do + {0, _} -> :ok + {count, _} -> {:ok, _} = Adapter.default_adapter().wake_sleeping_workflows(config, count) + end + end + + @doc """ + Returns the run's step executions, oldest first, including the steps its + parallel blocks ran. There is one row per attempt, and a step that waited + has one more row for the run after it resumed. + """ + @spec steps(WorkflowExecution.t() | map() | String.t(), keyword()) :: [StepExecution.t()] + def steps(run, opts \\ []) do + config = config(opts) + root = Repo.get(config, WorkflowExecution, run_id(run)) + + branches = + from(w in WorkflowExecution, + where: + w.parent_workflow_id == ^root.id and w.workflow_module == ^root.workflow_module and + w.workflow_name == ^root.workflow_name, + select: w.id + ) + + Repo.all( + config, + from(s in StepExecution, + where: s.workflow_id == ^root.id or s.workflow_id in subquery(branches), + order_by: [asc: s.inserted_at] + ) + ) + end + + @doc """ + Asserts the run's status, such as `:completed`, `:failed` or `:waiting`. On + a mismatch the failure shows the run's current step and error. + + Returns the reloaded run. + """ + @spec assert_status(WorkflowExecution.t() | map() | String.t(), atom(), keyword()) :: + WorkflowExecution.t() + def assert_status(run, expected, opts \\ []) do + run = Repo.get(config(opts), WorkflowExecution, run_id(run)) + + if run.status != expected do + flunk(""" + Expected workflow run #{run.id} to be #{inspect(expected)}, got #{inspect(run.status)} + current step: #{inspect(run.current_step)} + error: #{inspect(run.error, pretty: true)} + """) + end + + run + end + + @doc """ + Asserts that the step's latest attempt ended with `status` + (default `:completed`). Steps inside `parallel` and `branch` blocks are + matched by their own name, so it also checks which branch ran. + + Returns that `%StepExecution{}`, so its `output`, `error` and `attempt` can + be checked as well: + + assert assert_step(run, :charge).attempt == 3 + """ + @spec assert_step(WorkflowExecution.t() | map() | String.t(), atom(), atom(), keyword()) :: + StepExecution.t() + def assert_step(run, step, status \\ :completed, opts \\ []) when is_atom(status) do + all = steps(run, opts) + name = to_string(step) + + case all |> Enum.filter(&(short_name(&1) == name)) |> List.last() do + nil -> + ran = all |> Enum.map(&short_name/1) |> Enum.uniq() + flunk("Step #{inspect(step)} never ran. Steps that ran: #{inspect(ran)}") + + %StepExecution{status: ^status} = latest -> + latest + + latest -> + flunk(""" + Expected step #{inspect(step)} to be #{inspect(status)}, got #{inspect(latest.status)} \ + (attempt #{latest.attempt}) + error: #{inspect(latest.error, pretty: true)} + """) + end + end + + # Steps inside parallel and branch blocks are stored qualified, such as + # "parallel_123__email" or "branch_456__digital__ship". + defp short_name(%StepExecution{step_name: name}), + do: name |> String.split("__") |> List.last() + + @doc """ + Asserts that the run is waiting for the input named `input_name`. Returns the + `%PendingInput{}`. + """ + @spec assert_waiting_for_input( + WorkflowExecution.t() | map() | String.t(), + String.t(), + keyword() + ) :: + PendingInput.t() + def assert_waiting_for_input(run, input_name, opts \\ []) do + assert_pending(PendingInput, :input_name, run, input_name, opts) + end + + @doc """ + Asserts that the run is waiting for the event named `event_name`. Returns the + `%PendingEvent{}`. + """ + @spec assert_waiting_for_event( + WorkflowExecution.t() | map() | String.t(), + String.t(), + keyword() + ) :: + PendingEvent.t() + def assert_waiting_for_event(run, event_name, opts \\ []) do + assert_pending(PendingEvent, :event_name, run, event_name, opts) + end + + defp assert_pending(schema, field, run, name, opts) do + id = run_id(run) + + pending = + Repo.all( + config(opts), + from(p in schema, where: p.workflow_id == ^id and p.status == :pending) + ) + + Enum.find(pending, &(Map.fetch!(&1, field) == name)) || + flunk(""" + Expected workflow run #{id} to wait for #{inspect(name)} + it waits for: #{inspect(Enum.map(pending, &Map.fetch!(&1, field)))} + """) + end + + defp config(opts), do: Config.get(Keyword.get(opts, :durable, Durable)) + + defp run_id(%{id: id}), do: id + defp run_id(id) when is_binary(id), do: id +end diff --git a/durable/mix.exs b/durable/mix.exs index 202c675..420ff0f 100644 --- a/durable/mix.exs +++ b/durable/mix.exs @@ -102,6 +102,7 @@ defmodule Durable.MixProject do "guides/compensations.md", "guides/orchestration.md", "guides/parallel.md", + "guides/testing.md", "guides/waiting.md" ], groups_for_modules: [ diff --git a/durable/test/durable/testing_test.exs b/durable/test/durable/testing_test.exs new file mode 100644 index 0000000..4a1bf09 --- /dev/null +++ b/durable/test/durable/testing_test.exs @@ -0,0 +1,134 @@ +defmodule Durable.TestingTest do + use Durable.DataCase, async: false + + import Durable.Testing + + defmodule Order do + use Durable + use Durable.Helpers + use Durable.Wait + + workflow "order" do + step(:reserve, fn data -> {:ok, assign(data, :reserved, true)} end) + + parallel do + step(:email, fn data -> {:ok, assign(data, :emailed, true)} end) + step(:invoice, fn data -> {:ok, assign(data, :invoiced, true)} end) + end + + step(:cool_off, fn data -> + sleep(hours(24)) + {:ok, data} + end) + + step(:approve, fn data -> + %{"approved" => approved} = wait_for_input("approval") + {:ok, assign(data, :approved, approved)} + end) + end + end + + defmodule Flaky do + use Durable + use Durable.Helpers + + workflow "flaky" do + step(:charge, [retry: [max_attempts: 3]], fn data -> + calls = Process.get(:charge_calls, 0) + 1 + Process.put(:charge_calls, calls) + if calls < 3, do: {:error, "card declined"}, else: {:ok, assign(data, :charged, true)} + end) + end + end + + defmodule Payment do + use Durable + use Durable.Helpers + + workflow "payment" do + step(:authorize, fn _data -> {:ok, %{payment_id: "pay_1"}} end) + end + end + + defmodule Trip do + use Durable + use Durable.Helpers + use Durable.Orchestration + + workflow "trip" do + step(:pay, [compensate: :refund], fn data -> + {:ok, child} = call_workflow(Payment, %{}) + {:ok, assign(data, :payment_id, child["payment_id"])} + end) + + step(:book, fn _data -> {:error, "sold out"} end) + + compensate(:refund, fn data -> {:ok, data} end) + end + end + + defmodule Declined do + use Durable + + workflow "declined" do + step(:charge, fn _data -> {:error, "card declined"} end) + end + end + + test "string errors are stored as a map" do + run = run_workflow(Declined, %{}) + + assert_status(run, :failed) + assert run.error == %{"type" => "error", "message" => "card declined"} + end + + test "runs child workflows and compensations" do + run = run_workflow(Trip, %{}) + + assert_status(run, :compensated) + assert run.error == %{"type" => "error", "message" => "sold out"} + assert assert_step(run, :pay).status == :completed + assert_step(run, :book, :failed) + assert_step(run, :compensate_refund) + end + + test "drives parallel branches and sleeps, then stops at an input" do + run = run_workflow(Order, %{}) + + assert_status(run, :waiting) + assert_step(run, :email) + assert_step(run, :cool_off) + assert_waiting_for_input(run, "approval") + + :ok = Durable.provide_input(run.id, "approval", %{"approved" => true}) + run = run |> drain() |> assert_status(:completed) + + assert %{"approved" => true, "__results__" => %{"email" => ["ok", _]}} = run.context + end + + test "retries skip their backoff" do + {micros, run} = :timer.tc(fn -> run_workflow(Flaky, %{}) end) + + assert_status(run, :completed) + assert assert_step(run, :charge).attempt == 3 + # The default exponential backoff would wait 2s + 4s. + assert micros < 1_000_000 + end + + test "failures explain the mismatch" do + run = run_workflow(Order, %{}) + + error = assert_raise ExUnit.AssertionError, fn -> assert_status(run, :completed) end + assert error.message =~ "got :waiting" + assert error.message =~ ~s(current step: "approve") + + error = assert_raise ExUnit.AssertionError, fn -> assert_step(run, :ship) end + assert error.message =~ ~s(Step :ship never ran) + + error = assert_raise ExUnit.AssertionError, fn -> assert_step(run, :approve, :completed) end + assert error.message =~ "got :waiting" + + error = assert_raise ExUnit.AssertionError, fn -> assert_waiting_for_event(run, "paid") end + assert error.message =~ ~s(wait for "paid") + end +end diff --git a/site/src/docs.ts b/site/src/docs.ts index 843809c..acb5465 100644 --- a/site/src/docs.ts +++ b/site/src/docs.ts @@ -18,6 +18,7 @@ export const DOCS: { file: string; id: string; label: string; group: string; tit { file: 'durable/guides/compensations.md', id: 'docs/compensations', label: 'Compensations', group: 'Guides' }, { file: 'durable/guides/orchestration.md', id: 'docs/orchestration', label: 'Orchestration', group: 'Guides' }, { file: 'durable/guides/ai_workflows.md', id: 'docs/ai-workflows', label: 'AI workflows', group: 'Guides' }, + { file: 'durable/guides/testing.md', id: 'docs/testing', label: 'Testing', group: 'Guides' }, { file: 'durable_dashboard/README.md', id: 'docs/dashboard', label: 'Dashboard', group: 'Dashboard', title: 'Durable Dashboard' }, { file: 'durable/CHANGELOG.md', id: 'docs/changelog', label: 'durable', group: 'Changelog', description: 'Release history of durable, the workflow engine for Elixir: new features, fixes and breaking changes in every version on Hex.' }, { file: 'durable_dashboard/CHANGELOG.md', id: 'docs/changelog/dashboard', label: 'durable_dashboard', group: 'Changelog', description: 'Release history of durable_dashboard, the Phoenix LiveView dashboard for Durable: new pages, fixes and changes in every version on Hex.' }, From 0f83bfec37bddaa8752c98d81c279f5d7a4d59b9 Mon Sep 17 00:00:00 2001 From: Kasun Vithanage Date: Fri, 23 Oct 2026 07:21:10 +0530 Subject: [PATCH 3/3] docs: bring the roadmap up to date after 1.0 and the testing helpers --- agents/ROADMAP.md | 143 ++++++++---------- agents/milestones/M01-graph-visualization.md | 2 +- agents/milestones/M02-developer-tooling.md | 2 +- agents/milestones/M03-testing-framework.md | 22 ++- .../milestones/M04-documentation-release.md | 2 +- agents/milestones/M05-message-bus.md | 2 +- agents/milestones/M06-phoenix-dashboard.md | 2 +- agents/milestones/M07-scalability-adapters.md | 2 +- 8 files changed, 88 insertions(+), 89 deletions(-) diff --git a/agents/ROADMAP.md b/agents/ROADMAP.md index 9435daf..18f0d11 100644 --- a/agents/ROADMAP.md +++ b/agents/ROADMAP.md @@ -1,115 +1,94 @@ # Durable — Roadmap -**Last Updated:** 2026-03-09 +**Last Updated:** 2026-10-23 +**Current release:** `durable` 1.0.0 and `durable_dashboard` 1.0.0 on Hex (2026-10-21) **Reference:** See `arch.md` for technical architecture, `WORKPLAN_ARCHIVED.md` for historical detail --- ## Quick Stats -| Metric | Value | -|--------|-------| -| Source Modules | 42 | -| Passing Tests | ~291 | -| Documentation Guides | 6 | -| Lines of Code | ~11,000 | +| Metric | Core (`durable/`) | Dashboard (`durable_dashboard/`) | +|--------|-------------------|----------------------------------| +| Source modules | 77 | 37 | +| Tests | ~483 | ~235 | +| Lines of code | ~18,000 | — | +| Guides | 6 | — | --- ## Completed Work -| Area | Key Features | Tests | -|------|-------------|-------| -| **Foundation** | Mix project, Ecto schemas (6), programmatic migrations, NimbleOptions config, CI (`mix precommit`), Credo strict | — | -| **Core MVP** | `workflow`/`step`/`decision` macros, context management, executor, retry/backoff, PostgreSQL queue (FOR UPDATE SKIP LOCKED), query API, time helpers | ~65 | -| **Wait Primitives** | `sleep`, `schedule_at`, `wait_for_event`, `wait_for_any/all`, `wait_for_input`, `wait_for_approval`, `wait_for_choice`, `wait_for_text`, `wait_for_form`, timeouts | 52 | -| **Branching** | `branch on:` macro with pattern matching, default clause, multi-step branches | 19 | -| **Parallel Execution** | `parallel do` macro, `__results__` model, `into:` merge, `returns:` option, error strategies, resume durability | 20 | -| **Compensation/Saga** | `compensate` macro, `step :name, compensate:`, reverse-order execution | 10 | -| **Cron Scheduling** | `@schedule` decorator, scheduler GenServer, multi-node safety, timezone support, CRUD API, enable/disable, manual trigger | 49 | -| **Workflow Orchestration** | `call_workflow/3`, `start_workflow/3`, parallel call_workflow, cascade cancellation, parent notifications, `list_children/2` | 12 | -| **Observability** | Logger handler, IO capture, log buffer | 13 | -| **Developer Experience** | `mix durable.gen.migration`, `mix durable.install`, DataCase, @moduledoc/@doc/@spec, 6 guides | — | - -**Total Tests:** ~291 +| Area | Key Features | +|------|-------------| +| **Foundation** | Mix project, Ecto schemas, programmatic migrations, NimbleOptions config, CI (`mix precommit`), Credo strict | +| **Core MVP** | `workflow`/`step`/`decision` macros, context management, executor, retry/backoff, PostgreSQL queue (FOR UPDATE SKIP LOCKED), query API, time helpers | +| **Wait Primitives** | `sleep`, `schedule_at`, `wait_for_event`, `wait_for_any/all`, `wait_for_input`, `wait_for_approval`, `wait_for_choice`, `wait_for_text`, `wait_for_form`, timeouts | +| **Branching** | `branch on:` macro with pattern matching, default clause, multi-step branches | +| **Parallel Execution** | `parallel do` macro, `__results__` model, `into:` merge, `returns:` option, error strategies, resume durability, deterministic block ids | +| **Compensation/Saga** | `compensate` macro, `step :name, compensate:`, reverse-order execution | +| **Cron Scheduling** | `@schedule` decorator, scheduler GenServer, multi-node safety, timezone support, CRUD API, enable/disable, manual trigger, failure tracking + auto-disable, `upcoming_runs/2` | +| **Workflow Orchestration** | `call_workflow/3`, `start_workflow/3`, parallel call_workflow, cascade cancellation, parent notifications, `list_children/2` | +| **Hardening (Apr 2026 audit)** | JSONB sanitization at every write boundary, context survives retries/gotos, `Ecto.Multi` for timeout/child/wait-group transitions, ack retries, `:compensating` zombie recovery — see `docs/bug-reports/2026-04-13-follow-up-audit.md` | +| **Operator Controls** | Retry/rerun, cancel (CAS race fixed), queue pause, workflow `timeout:` (deadline enforced between steps), workflow `queue:`, `trigger` tracking, pruner, node registry | +| **Observability** | Logger handler, IO capture, log buffer, `Durable.Telemetry` events + metrics, `Durable.Stats`, `Durable.Activity` log, queue sampler, `Durable.PubSub` lifecycle broadcasts | +| **Testing ([M03](milestones/M03-testing-framework.md))** | `Durable.Testing`: `run_workflow`, `drain` (runs parallel branches + child workflows, skips sleeps and retry backoff), `assert_status`/`assert_step`/`assert_waiting_for_input`/`assert_waiting_for_event`, `guides/testing.md` | +| **CLI ([M02](milestones/M02-developer-tooling.md))** | `mix durable.*`: status, list, inspect, run, cancel, retry, cleanup, pending, provide_input, send_event, doctor, install, migrations, gen.migration, gen.upgrade | +| **Dashboard ([M06](milestones/M06-phoenix-dashboard.md))** | Separate `durable_dashboard` package: LiveView pages (Overview, Runs, Run, Workflows, Inbox, Queues, Schedules, Telemetry, Settings), ReactFlow graph island, ⌘K palette | +| **Docs & Release ([M04](milestones/M04-documentation-release.md))** | HexDocs, 6 guides, git-cliff changelogs, GitHub Actions release + Hex publish, durable.wavezync.com, exported `locals_without_parens` for the DSL (#10) | --- -## Active Milestones - -| ID | Title | Priority | Effort | Status | Dependencies | -|----|-------|----------|--------|--------|--------------| -| [M01](milestones/M01-graph-visualization.md) | Graph Visualization | Medium | 1 week | Not Started | None | -| [M02](milestones/M02-developer-tooling.md) | Developer Tooling | High | 1 week | Not Started | None | -| [M03](milestones/M03-testing-framework.md) | Testing Framework | Medium | 1 week | Not Started | None | -| [M04](milestones/M04-documentation-release.md) | Documentation & Release | High | 1 week | Not Started | M02, M03 (soft) | -| [M05](milestones/M05-message-bus.md) | Message Bus | Medium | 1.5 weeks | Not Started | None | -| [M06](milestones/M06-phoenix-dashboard.md) | Phoenix Dashboard | Low | 2 weeks | Not Started | M01, M05 | -| [M07](milestones/M07-scalability-adapters.md) | Scalability Adapters | Low | 2 weeks | Not Started | M05 (soft) | - -### Recommended Execution Order - -``` -M02 (Developer Tooling) ─┐ -M03 (Testing Framework) ─┤──→ M04 (Documentation & Release) -M01 (Graph Visualization) ┘ - ──→ M05 (Message Bus) - ├──→ M06 (Phoenix Dashboard) - └──→ M07 (Scalability Adapters) -``` - -**Phase A** (parallel, no dependencies): M01, M02, M03 -**Phase B** (needs Phase A): M04 -**Phase C** (independent): M05 -**Phase D** (needs M01 + M05): M06, M07 +## Milestones + +| ID | Title | Status | Notes | +|----|-------|--------|-------| +| [M01](milestones/M01-graph-visualization.md) | Graph Visualization | Superseded | Dashboard `GraphBuilder` + ReactFlow covers it. Core DOT/Mermaid export not built → backlog | +| [M02](milestones/M02-developer-tooling.md) | Developer Tooling | Done | Shipped more tasks than planned | +| [M03](milestones/M03-testing-framework.md) | Testing Framework | Done | Shipped as `Durable.Testing` helpers (imported into the user's own case) rather than a `Durable.TestCase` template | +| [M04](milestones/M04-documentation-release.md) | Documentation & Release | Done | 1.0.0 published | +| [M05](milestones/M05-message-bus.md) | Message Bus | Partial | `Durable.PubSub` lifecycle events shipped. Adapter behaviour / pg_notify not built → backlog | +| [M06](milestones/M06-phoenix-dashboard.md) | Phoenix Dashboard | Done | Shipped as its own package | +| [M07](milestones/M07-scalability-adapters.md) | Scalability Adapters | Backlog | No demand for Redis yet | --- -## Backlog +## Next Up + +In priority order: + +1. **Correctness follow-ups from v1:** + - The workflow deadline can't interrupt a step that's still running; it's only checked between steps. + - `wait_for_input` with `on_timeout: :fail` ends the run as `cancelled`, not `failed`. Fixing it changes behaviour, so ship it in a minor release with a changelog note. +2. **Workflow versioning.** Needs a milestone doc. Changing a deployed workflow while runs are still in flight is the main open problem for production users. +3. **DSL gaps:** waits inside `parallel` blocks, an `each`/foreach primitive (each iteration needs its own context, see `Durable.DSL.Step` docs), decisions inside `parallel`. +4. **Testing follow-up:** wait timeouts can't be exercised by `Durable.Testing.drain/2` (the timeout worker only runs with the queue on). A helper that fires a run's pending timeouts would close it. +5. **Formatter follow-up to #10:** the repo's own `durable/.formatter.exs` only *exports* `locals_without_parens`, so internal code and tests still get parens. Also add `:durable` to `import_deps` in `durable_dashboard` and `examples/phoenix_demo`. + +--- -Lower-priority items not warranting standalone milestone documents: +## Backlog | Item | Notes | |------|-------| +| DOT/Mermaid graph export | Leftover from M01. For docs/READMEs, if requested | +| Message bus adapter behaviour / pg_notify | Leftover from M05. `Phoenix.PubSub` already spans nodes | +| Redis queue adapter | M07. Add when someone needs it | +| RabbitMQ queue adapter | Niche — add if requested | | Switch/Case macro | Low priority — `branch` covers most use cases | | Pipe-based API | Functional workflow composition — nice-to-have | -| Example Project | Standalone demo app with common patterns | -| Redis message bus | Quick follow-up after M05 + M07 | -| RabbitMQ queue adapter | Niche — add if requested | - ---- - -## Test Coverage - -| Test File | Tests | Area | -|-----------|-------|------| -| wait_test.exs | 52 | Wait primitives | -| scheduler_test.exs | 49 | Cron scheduling | -| parallel_test.exs | 20 | Parallel execution | -| branch_test.exs | 19 | Branch macro | -| postgres_test.exs | 16 | Queue adapter | -| decision_test.exs | 14 | Decision steps | -| log_capture_test.exs | 13 | Log/IO capture | -| orchestration_test.exs | 12 | Workflow orchestration | -| integration_test.exs | 11 | End-to-end flows | -| validation_test.exs | 10 | Input validation | -| context_test.exs | 10 | Context management | -| compensation_test.exs | 10 | Saga pattern | -| durable_test.exs | 10 | Core API | -| handler_test.exs | 8 | Log handler | -| io_server_test.exs | 7 | IO capture | -| resume_edge_cases_test.exs | 5 | Resume edge cases | -| log_capture/integration_test.exs | 5 | Log capture integration | -| Other | ~20 | Misc | -| **Total** | **~291** | | --- ## Known Limitations -1. Wait primitives not supported in parallel blocks -2. Child workflows with waits (`sleep`, `wait_for_event`) not supported in parallel blocks +1. Wait primitives (and child workflows that wait) are not supported inside parallel blocks +2. Decisions are not supported inside parallel blocks 3. No backward jumps in decision steps (forward-only by design) 4. Context is single-level atomized (top-level keys only) 5. No workflow versioning -6. No foreach/loop DSL primitives (use Elixir's `Enum` functions) +6. No foreach/loop DSL primitive (use Elixir's `Enum` functions) +7. A resumed wait re-runs the step body from the top; side effects before the wait must be idempotent +8. Only one `sleep/1` per step body +9. The workflow deadline is checked between steps and can't interrupt a running step +10. `Durable.Testing.drain/2` doesn't fire wait timeouts diff --git a/agents/milestones/M01-graph-visualization.md b/agents/milestones/M01-graph-visualization.md index 251f646..5a0fb0e 100644 --- a/agents/milestones/M01-graph-visualization.md +++ b/agents/milestones/M01-graph-visualization.md @@ -1,6 +1,6 @@ # M01: Graph Visualization -**Status:** Not Started +**Status:** Superseded (dashboard GraphBuilder; DOT/Mermaid export moved to backlog) **Priority:** Medium **Effort:** 1 week **Dependencies:** None diff --git a/agents/milestones/M02-developer-tooling.md b/agents/milestones/M02-developer-tooling.md index 406c260..3aceee2 100644 --- a/agents/milestones/M02-developer-tooling.md +++ b/agents/milestones/M02-developer-tooling.md @@ -1,6 +1,6 @@ # M02: Developer Tooling -**Status:** Not Started +**Status:** Done **Priority:** High **Effort:** 1 week **Dependencies:** None diff --git a/agents/milestones/M03-testing-framework.md b/agents/milestones/M03-testing-framework.md index db8caa6..b7dc882 100644 --- a/agents/milestones/M03-testing-framework.md +++ b/agents/milestones/M03-testing-framework.md @@ -1,6 +1,6 @@ # M03: Testing Framework -**Status:** Not Started +**Status:** Done (shipped as `Durable.Testing` + `guides/testing.md`; see note below) **Priority:** Medium **Effort:** 1 week **Dependencies:** None @@ -175,6 +175,26 @@ Sections: - [ ] Existing tests in `test/` still pass (no regressions) - [ ] `mix credo --strict` passes +## Outcome (2026-10-23) + +Shipped as `Durable.Testing`, a helper module users `import` into their own +`DataCase`, instead of a `Durable.TestCase` case template: users already have a +case template that owns the sandbox, and ExUnit allows only one per module. + +- `run_workflow/3`, `drain/2`: inline execution of the whole run tree (parallel + branches, child workflows); sleeps end immediately, retry backoff is skipped + (`Backoff.pause/1` checks `:durable_skip_backoff`). +- `assert_status/3`, `assert_step/4` (matches parallel/branch steps by their own + name, returns the `%StepExecution{}`), `assert_waiting_for_input/3`, + `assert_waiting_for_event/3`, `steps/2`. +- Dropped as unneeded: `assert_context` (pattern-match `run.context`), + `assert_completed_within`, `provide_test_input`/`send_test_event` (use + `Durable.provide_input/3` / `send_event/3` + `drain/2`). +- Not covered: wait timeouts (TimeoutWorker needs the queue on). +- Found and fixed while testing it: a step returning `{:error, "string"}` left + the run stuck `:running` (or crashed compensation) because the error wasn't + normalized to a map before the write. + ## Open Questions - Should `run_workflow_sync` return a rich struct or the raw `%WorkflowExecution{}` schema? diff --git a/agents/milestones/M04-documentation-release.md b/agents/milestones/M04-documentation-release.md index f98f6a3..9f91cd4 100644 --- a/agents/milestones/M04-documentation-release.md +++ b/agents/milestones/M04-documentation-release.md @@ -1,6 +1,6 @@ # M04: Documentation & Release -**Status:** Not Started +**Status:** Done **Priority:** High **Effort:** 1 week **Dependencies:** M02, M03 (soft — can start without, but references their output) diff --git a/agents/milestones/M05-message-bus.md b/agents/milestones/M05-message-bus.md index 35f4c62..3f107df 100644 --- a/agents/milestones/M05-message-bus.md +++ b/agents/milestones/M05-message-bus.md @@ -1,6 +1,6 @@ # M05: Message Bus -**Status:** Not Started +**Status:** Partial (Durable.PubSub lifecycle events; adapter behaviour moved to backlog) **Priority:** Medium **Effort:** 1.5 weeks **Dependencies:** None diff --git a/agents/milestones/M06-phoenix-dashboard.md b/agents/milestones/M06-phoenix-dashboard.md index 87c4a47..4ce1d38 100644 --- a/agents/milestones/M06-phoenix-dashboard.md +++ b/agents/milestones/M06-phoenix-dashboard.md @@ -1,6 +1,6 @@ # M06: Phoenix Dashboard -**Status:** Not Started +**Status:** Done **Priority:** Low **Effort:** 2 weeks **Dependencies:** M01 (graph visualization), M05 (message bus for real-time updates) diff --git a/agents/milestones/M07-scalability-adapters.md b/agents/milestones/M07-scalability-adapters.md index 1ffa927..bf483a9 100644 --- a/agents/milestones/M07-scalability-adapters.md +++ b/agents/milestones/M07-scalability-adapters.md @@ -1,6 +1,6 @@ # M07: Scalability Adapters -**Status:** Not Started +**Status:** Backlog **Priority:** Low **Effort:** 2 weeks **Dependencies:** M05 (soft — message bus pattern informs adapter design)