Cracow decisions on the DomovoyCore workflow

What this notebook shows

The engine notebook priced one flat with one stage: a diamond-shaped graph run by DomovoyCore.Engine, remembered by the store and the journal. This notebook answers the three bullets of its last section, in order.

• A person joins the run. The stage moves into a DomovoyCore.Workflow, after a DomovoyCore.Decision with the default decider, DomovoyCore.Decider.Person. The run stops, the person reads the estimate and either accepts it or sends the stage back for a re-run at the next generation. DomovoyCore.Run folds the journal to know where it is, and DomovoyCore.Workflow.Server owns the run as a process.
• A branch is swapped without touching the graph. runners: replaces the location model at execution time, and the engine checks the replacement against the declared input schema before anything runs.
• A flaky branch retries, on a budget. One node gets an exponential, jittered retry:, its runner fails twice with a retryable error and succeeds on the third attempt, and the store keeps every attempt. A second run exhausts max_attempts, a third runs out of its deadline_ms, and a fourth trips the breaker of a runner that two branches share.

This notebook stands alone: it rebuilds the estimate stage it needs, so evaluate it top to bottom on its own. If the engine notebook ran first in the same server, the cells below print redefinition warnings on the shared attached node; the new definitions are identical, and everything still runs.

Every coefficient here is illustrative, as before. Nothing is fitted to real Kraków listings.

Start a runtime

The same host as in the engine notebook: a named DomovoyCore.Runtime owned by the notebook, which every run names.

case DomovoyCore.Runtime.start_link(name: Cracow.Runtime) do
  {:ok, _pid} -> :ok
  {:error, {:already_started, _pid}} -> :ok
end

alias DomovoyCore.{Choice, Context, Decision, Engine, Graph, Job, Journal, Node, Run, Stage, Store, Type, Value, Workflow}
alias DomovoyCore.Workflow.Server

The estimate stage, again

The same five runners as in the engine notebook, compressed to what this notebook needs: a float type the engine does not ship, the illustrative price model, one validator, and the diamond. Read the engine notebook for why each piece looks this way; here they are the stage the workflow runs.

defmodule Cracow.Type.Float do
  @moduledoc "A `DomovoyCore.Type` for a float, such as an area in square metres."
  use DomovoyCore.Type

  @impl Ecto.Type
  def type, do: :float

  @impl DomovoyCore.Type
  def cast(raw, _metadata) when is_float(raw), do: {:ok, raw}
  def cast(raw, _metadata) when is_integer(raw), do: {:ok, raw * 1.0}
  def cast(_raw, _metadata), do: :error
end

defmodule Cracow.Market do
  @moduledoc "An illustrative price model for flats in Kraków."

  @districts %{
    "Stare Miasto" => 1.35,
    "Kazimierz" => 1.30,
    "Krowodrza" => 1.12,
    "Dębniki" => 1.08,
    "Podgórze" => 1.05,
    "Bronowice" => 1.02,
    "Prądnik Biały" => 0.95,
    "Bieżanów-Prokocim" => 0.88,
    "Nowa Huta" => 0.80
  }

  def districts, do: @districts |> Map.keys() |> Enum.sort()
  def district_factor(district), do: Map.fetch!(@districts, district)

  def base_value(area), do: area * (17_800 - 42 * area + 0.16 * area * area)

  def layout_factor(rooms, bathrooms, area) do
    crowding = max(0.0, rooms - area / 18)
    1 + 0.025 * (rooms - 1) + 0.04 * (bathrooms - 1) - 0.02 * crowding * crowding
  end

  def location_factor(district, floor, balcony?, year_built) do
    district_factor(district) * floor_factor(floor) * balcony_factor(balcony?) *
      age_factor(year_built)
  end

  defp floor_factor(0), do: 0.96
  defp floor_factor(floor) when floor >= 5, do: 1.03
  defp floor_factor(_floor), do: 1.0

  defp balcony_factor(true), do: 1.03
  defp balcony_factor(false), do: 1.0

  defp age_factor(year) when year >= 2015, do: 1.08
  defp age_factor(year) when year < 1960, do: 1.02
  defp age_factor(year) when year <= 1989, do: 0.94
  defp age_factor(_year), do: 1.0
end

defmodule Cracow.Validator.Listing do
  @moduledoc "Refuses a listing no flat in Kraków could match."
  @behaviour DomovoyCore.Validator
  import Ecto.Changeset

  @impl DomovoyCore.Validator
  def validate(changeset, %DomovoyCore.Context{}) do
    changeset
    |> validate_number(:rooms, greater_than_or_equal_to: 1)
    |> validate_number(:bathrooms, greater_than_or_equal_to: 1)
    |> validate_number(:area, greater_than: 0)
    |> validate_number(:floor, greater_than_or_equal_to: 0)
    |> validate_inclusion(:district, Cracow.Market.districts())
  end
end

defmodule Cracow.Runner.Normalize do
  @moduledoc "Checks the listing and gives one feature map to every branch."
  use DomovoyCore.Runner

  input do
    field :rooms, DomovoyCore.Type.Integer
    field :bathrooms, DomovoyCore.Type.Integer
    field :area, Cracow.Type.Float
    field :district, DomovoyCore.Type.String
    field :floor, DomovoyCore.Type.Integer, default: 1
    field :balcony, DomovoyCore.Type.Boolean, default: false
    field :year_built, DomovoyCore.Type.Integer, default: 2005
  end

  required [:rooms, :bathrooms, :area, :district]
  validators [Cracow.Validator.Listing]

  @impl DomovoyCore.Runner
  def run(%Input{} = listing, %DomovoyCore.Context{}) do
    features =
      listing
      |> Map.from_struct()
      |> Map.new(fn {field, value} -> {Atom.to_string(field), value} end)
      |> Map.put("m2_per_room", listing.area / listing.rooms)

    {:ok, features}
  end
end

defmodule Cracow.Runner.AreaModel do
  @moduledoc "The base value of the flat, from its area alone."
  use DomovoyCore.Runner

  input do
    field :features, DomovoyCore.Type.Map
  end

  required [:features]

  @impl DomovoyCore.Runner
  def run(%Input{features: features}, %DomovoyCore.Context{}) do
    {:ok, Cracow.Market.base_value(features["area"])}
  end
end

defmodule Cracow.Runner.LayoutModel do
  @moduledoc "The rooms and bathrooms factor."
  use DomovoyCore.Runner

  input do
    field :features, DomovoyCore.Type.Map
  end

  required [:features]

  @impl DomovoyCore.Runner
  def run(%Input{features: features}, %DomovoyCore.Context{}) do
    %{"rooms" => rooms, "bathrooms" => bathrooms, "area" => area} = features
    {:ok, Cracow.Market.layout_factor(rooms, bathrooms, area)}
  end
end

defmodule Cracow.Runner.LocationModel do
  @moduledoc "The district, floor, balcony and age factor."
  use DomovoyCore.Runner

  input do
    field :features, DomovoyCore.Type.Map
  end

  required [:features]

  @impl DomovoyCore.Runner
  def run(%Input{features: features}, %DomovoyCore.Context{}) do
    %{"district" => district, "floor" => floor, "balcony" => balcony, "year_built" => year} =
      features

    {:ok, Cracow.Market.location_factor(district, floor, balcony, year)}
  end
end

defmodule Cracow.Runner.Estimate do
  @moduledoc "Joins the three terms into one price."
  use DomovoyCore.Runner

  input do
    field :features, DomovoyCore.Type.Map
    field :base, Cracow.Type.Float
    field :layout, Cracow.Type.Float
    field :location, Cracow.Type.Float
  end

  required [:features, :base, :layout, :location]

  @impl DomovoyCore.Runner
  def run(%Input{} = input, %DomovoyCore.Context{}) do
    price = input.base * input.layout * input.location

    {:ok,
     %{
       "price" => round(price),
       "per_m2" => round(price / input.features["area"]),
       "components" => %{
         "base" => round(input.base),
         "layout" => input.layout,
         "location" => input.location
       }
     }}
  end
end

The graph is the same diamond, and the stage now names what runs next: the decision. A stage has one control successor; a stage that branches points at a decision.

graph =
  Graph.new([
    Node.new(%{
      name: "normalize",
      runner: Cracow.Runner.Normalize,
      type: Type.Map,
      bind: %{
        rooms: {"rooms", Type.Integer},
        bathrooms: {"bathrooms", Type.Integer},
        area: {"area", Cracow.Type.Float},
        district: {"district", Type.String},
        floor: {"floor", Type.Integer},
        balcony: {"balcony", Type.Boolean},
        year_built: {"year_built", Type.Integer}
      }
    }),
    Node.new(%{
      name: "area_model",
      runner: Cracow.Runner.AreaModel,
      type: Cracow.Type.Float,
      bind: %{features: {"normalize", Type.Map}}
    }),
    Node.new(%{
      name: "layout_model",
      runner: Cracow.Runner.LayoutModel,
      type: Cracow.Type.Float,
      bind: %{features: {"normalize", Type.Map}}
    }),
    Node.new(%{
      name: "location_model",
      runner: Cracow.Runner.LocationModel,
      type: Cracow.Type.Float,
      bind: %{features: {"normalize", Type.Map}}
    }),
    Node.new(%{
      name: "estimate",
      runner: Cracow.Runner.Estimate,
      type: Type.Map,
      bind: %{
        features: {"normalize", Type.Map},
        base: {"area_model", Cracow.Type.Float},
        layout: {"layout_model", Cracow.Type.Float},
        location: {"location_model", Cracow.Type.Float}
      }
    })
  ])

estimate_stage = Stage.new(%{name: "estimate_price", graph: graph, next: "review_estimate"})

The workflow: a stage, then a person

The decision asks what happens next. It runs no runner and declares no bindings; the store holds what the person reads. Two choices: take the estimate and end the workflow, or correct the district and run the stage again. The revise choice declares one typed input, so the person can send a new district with the answer, and the record of that district lands in the store at the new generation.

review =
  Decision.new(%{
    name: "review_estimate",
    prompt: "A flat in Kraków was priced. Accept the estimate, or send the listing back with a corrected district?",
    choices: [
      Choice.new(%{
        name: "accept",
        description: "Take the estimate. The workflow ends.",
        target: :halt
      }),
      Choice.new(%{
        name: "revise",
        description: "Correct the district. The stage runs again at the next generation.",
        target: {:rerun, "estimate_price"},
        inputs: %{"district" => Type.String}
      })
    ]
  })

The workflow holds the two vertices, the start, and the listing fields a run begins with. Data travels through the store, not on the arrows: the seven inputs are records the stage reads, and the choice input district writes a record under the same name a workflow input wrote. The file adapters keep every run under its own directory, cleared here so the sections below start empty.

root = Path.join(System.tmp_dir!(), "domovoy_cracow_sale")
File.rm_rf!(root)

workflow =
  Workflow.new!(%{
    name: "cracow_sale",
    vertices: %{"estimate_price" => estimate_stage, "review_estimate" => review},
    start: "estimate_price",
    inputs: %{
      "rooms" => %{type: Type.Integer},
      "bathrooms" => %{type: Type.Integer},
      "area" => %{type: Cracow.Type.Float},
      "district" => %{type: Type.String},
      "floor" => %{type: Type.Integer, default: 1},
      "balcony" => %{type: Type.Boolean, default: false},
      "year_built" => %{type: Type.Integer, default: 2005}
    },
    store: {Store.FileSystem, [root: root]},
    journal: {Journal.FileSystem, [root: root]}
  })

Workflow.new!/1 checked the whole shape before anything runs: every graph input of the stage is a workflow input, the re-run points at a stage, and no node name collides with an input. The control picture is two arrows; only a decision branches, and :halt draws none.

flowchart LR
  estimate_price[estimate_price]
  review_estimate{review_estimate}
  halt((halt))
  estimate_price --> review_estimate
  review_estimate -->|revise| estimate_price
  review_estimate -->|accept| halt

The Elixir cell regenerates the arrows from the workflow, so the picture and the workflow cannot drift apart. The generated source carries no labels — the static block above adds the choice names by hand — but every arrow comes from workflow.arrows.

for arrow <- workflow.arrows, do: "  #{arrow.from} --> #{arrow.to}"

A few helpers for the cells that follow: the estimate out of the store, and small tables.

defmodule Cracow.Sale do
  @moduledoc "Glue between the notebook cells and the workflow."

  alias DomovoyCore.{Journal, Record, Store, Value}

  @doc "The value of the `estimate` node at `generation`."
  def estimate(store, generation) do
    {:ok, %Record{result: %Value{value: value}}} = Store.get(store, "estimate", generation)
    value
  end

  @doc "The kinds of every journaled event, in order."
  def event_kinds(journal) do
    {:ok, events} = Journal.events(journal)
    Enum.map(events, & &1.kind)
  end

  @doc "One row per record: where it sits in the store and what it holds."
  def records_table(records) when is_list(records) do
    for record <- records do
      %{
        node: record.node,
        generation: record.job.generation,
        attempt: record.job.attempt,
        status: record.status,
        result: describe(record.result)
      }
    end
  end

  def records_table(records) when is_map(records),
    do: records |> Map.values() |> Enum.sort_by(& &1.node) |> records_table()

  @doc "Formats a PLN amount with thousands separators."
  def pln(amount) when is_number(amount) do
    amount
    |> round()
    |> Integer.to_string()
    |> String.reverse()
    |> String.replace(~r/(\d{3})(?=\d)/, "\\1 ")
    |> String.reverse()
    |> Kernel.<>(" PLN")
  end

  defp describe(%Value{value: value}) when is_map(value), do: inspect(value, limit: 6)
  defp describe(%Value{value: value}), do: inspect(value)
  defp describe(%DomovoyCore.Error{type: type}), do: "error: #{type}"
  defp describe(nil), do: ""
end

Run it to the decision

Run.start/4 casts the raw listing through the declared workflow inputs, writes one record per input, appends :run_started and puts the cursor on start. run_to_decision/2 steps until something waits for a person: the stage runs its five runners, the cursor reaches the review, and the default decider answers :await.

listing = %{
  "rooms" => 3,
  "bathrooms" => 1,
  "area" => 58.5,
  "district" => "Krowodrza",
  "floor" => 3,
  "balcony" => false,
  "year_built" => 2010
}

run = Run.start(Cracow.Runtime, workflow, listing, job: Job.new("sale-1"))
run = Run.run_to_decision(workflow, run)
{run.status, run.cursor, run.job.generation}

The run is small: a status, a cursor and a job. The records hold the weight. This cell is the person's screen: the prompt, the offered choices with what each one does, and the estimate the store holds at generation 0.

decision = Workflow.vertex(workflow, run.cursor)

{
  decision.prompt,
  Enum.map(decision.choices, &%{choice: &1.name, does: &1.description}),
  Cracow.Sale.pln(Cracow.Sale.estimate(run.store, 0)["price"])
}

The journal of the run tells the same story as events: the run began, the stage began, five nodes started and finished, the stage finished, and the decision waits.

Cracow.Sale.event_kinds(run.journal)

The person sends it back

A wrong choice refuses softly: the run keeps waiting with the error set, so the person can answer again. Nothing is journaled for a refusal.

refused = Run.decide(workflow, run, "maybe", %{})
{refused.status, refused.cursor, refused.error.type}

The district is wrong — the flat is in Stare Miasto, not Krowodrza — so the person revises and sends the corrected district with the answer. The answer is recorded under the decision, the generation rises, the district lands in the store at generation 1, the :decided event is journaled, and the cursor is back on the stage.

run = Run.decide(workflow, run, "revise", %{"district" => "Stare Miasto"})
{run.status, run.cursor, run.job.generation}

The stage runs again, and waits again. Generation 0 still holds the first estimate; generation 1 holds the second. The district factor rose from 1.12 to 1.35, and the price follows it.

run = Run.run_to_decision(workflow, run)
{run.status, run.cursor, run.job.generation}

gen0 = Cracow.Sale.estimate(run.store, 0)["price"]
gen1 = Cracow.Sale.estimate(run.store, 1)["price"]
{Cracow.Sale.pln(gen0), Cracow.Sale.pln(gen1)}

The journal knows where the run is

The run above waits with generation 1, but that state lives in one variable of this notebook. Run.replay/4 rebuilds the same state from the journal alone: it opens the store and the journal by run id, folds every event with the pure Run.apply/2, and hands back a run a new process can continue. The cursor, the status and the generation come from the events, not from memory.

replayed = Run.replay(Cracow.Runtime, workflow, "sale-1")
{replayed.status, replayed.cursor, replayed.job.generation}

The person accepts the second estimate. The choice halts the workflow, so the last event is :run_halted and the cursor is nil: there is nowhere left to go.

run = Run.decide(workflow, run, "accept", %{})
{run.status, run.cursor, List.last(Cracow.Sale.event_kinds(run.journal))}

A process owns the run

Run is data a caller drives. DomovoyCore.Workflow.Server is a process that drives itself: it holds one run, executes the start and every step in tasks under the runtime, and stays responsive while they run. state/1 gives a view — workflow, run id, status, cursor, generation, error — and never the open store or journal.

Subscribe first, then start: the journal broadcasts every event on the run's topic, and the mailbox below collects them while the server works. A second start/4 with the same run id, workflow name and inputs gives the same pid instead of a second run.

server_job = Job.new("sale-server-1")
:ok = Server.subscribe(Cracow.Runtime, workflow.name, server_job.id)
{:ok, server} = Server.start(Cracow.Runtime, workflow, listing, job: server_job)
is_pid(server)

The server drives in the background, so this cell waits for the state it wants instead of assuming it. state/1 answers {:error, :starting} while initialization runs; then the stage runs; then the run waits.

wait_for = fn server, status ->
  Enum.reduce_while(1..200, nil, fn n, _ ->
    case Server.state(server) do
      %{status: ^status} = view -> {:halt, view}
      _ when n < 200 -> Process.sleep(50); {:cont, nil}
      other -> {:halt, {:timeout, other}}
    end
  end)
end

awaiting = wait_for.(server, :awaiting_decision)
{awaiting.status, awaiting.cursor, awaiting.generation}

The :decision_awaited event arrived over PubSub while the server drove, like in the engine notebook — the server sends no broadcasts of its own, the journal append does.

receive do
  {:domovoy_event, %DomovoyCore.Event{kind: :decision_awaited}} -> :awaited
after
  10_000 -> :timeout
end

The person answers through the server. decide/3 applies Run.decide/4 in the server process and gives the post-decide view; the background drive then runs the stage at generation 1 while the caller goes on. Answering while a step still runs gives run_busy instead — the wait above is what keeps this cell out of that race.

{:ok, decided} = Server.decide(server, "revise", %{"district" => "Kazimierz"})
{decided.status, decided.cursor, decided.generation}

Wait for the second review, accept through the server, and read the final view. Halting journals :run_halted inside the decide call, so the view is already finished.

awaiting_again = wait_for.(server, :awaiting_decision)
{awaiting_again.status, awaiting_again.cursor, awaiting_again.generation}
{:ok, accepted} = Server.decide(server, "accept", %{})
{accepted.status, accepted.cursor}
Server.state(server)

Stopping the server changes nothing on disk: the store and the journal stay, keyed by workflow name and run id. resume/4 folds the journal back into a run and starts a fresh owner for it — a cold owner replays and carries on. The resumed view below is the finished run, rebuilt from events.

:ok = Server.stop(server)
{:ok, resumed} = Server.resume(Cracow.Runtime, workflow, "sale-server-1")
Server.state(resumed)

Leave no processes behind for the next section.

:ok = Server.stop(resumed)

Swap a branch without touching the graph

The location model is one module behind the name "location_model". The runners: option replaces the module at execution time — on Engine.run/4, and through it on Run.start/4, which validates every override before any stage runs and keeps them through decisions and generations. This replacement declares the same input schema, a features map, and computes a forecast: the location factor with a 10% premium.

defmodule Cracow.Runner.LocationModelPremium do
  @moduledoc "A forecast: the location factor with a 10% premium."
  use DomovoyCore.Runner

  input do
    field :features, DomovoyCore.Type.Map
  end

  required [:features]

  @impl DomovoyCore.Runner
  def run(%Input{features: features}, %DomovoyCore.Context{}) do
    %{"district" => district, "floor" => floor, "balcony" => balcony, "year_built" => year} =
      features

    {:ok, Cracow.Market.location_factor(district, floor, balcony, year) * 1.10}
  end
end

The graph, the workflow and the listing are untouched; only the execution changes. The estimate comes back exactly 10% above the standard one, under the same node names.

premium_run =
  Run.start(Cracow.Runtime, workflow, listing,
    job: Job.new("sale-premium-1"),
    runners: %{"location_model" => Cracow.Runner.LocationModelPremium}
  )

premium_run = Run.run_to_decision(workflow, premium_run)
{premium_run.status, premium_run.cursor}

standard = Cracow.Sale.estimate(run.store, 0)["price"]
premium = Cracow.Sale.estimate(premium_run.store, 0)["price"]
{standard, premium, Float.round(premium / standard, 3)}

The override survives the person's answer: deciding keeps the replacement for the re-run, because the run carries runners and Run revalidates nothing mid-flight.

premium_run = Run.decide(workflow, premium_run, "revise", %{"district" => "Kazimierz"})
{premium_run.status, premium_run.cursor, premium_run.job.generation}
premium_run.runners

The check runs before anything else. This replacement declares features as a string, so its schema is incompatible with the declared runner, and the run fails before it opens a store: status is :failed, and there are no records and no journal — validation runs before anything is written.

defmodule Cracow.Runner.LocationModelBroken do
  @moduledoc "A replacement with an incompatible input schema."
  use DomovoyCore.Runner

  input do
    field :features, DomovoyCore.Type.String
  end

  required [:features]

  @impl DomovoyCore.Runner
  def run(%Input{}, %DomovoyCore.Context{}), do: {:ok, 1.0}
end

broken =
  Run.start(Cracow.Runtime, workflow, listing,
    job: Job.new("sale-broken-1"),
    runners: %{"location_model" => Cracow.Runner.LocationModelBroken}
  )

{broken.status, broken.error.type, broken.error.reason}

A flaky branch retries

Back down a level, to one node of the graph. Node.new/1 takes retry:, a DomovoyCore.Retry: max_attempts, a backoff, its jitter, a timeout_ms per attempt and, further down, a deadline_ms per node and a breaker per runner. This cell gives the location model an exponential backoff on the same graph, by replacing the retry of one node — the graph construction above is untouched.

alias DomovoyCore.Retry

flaky_retry =
  Retry.new(
    max_attempts: 3,
    backoff: {:exponential, 250, 2, 2_000},
    jitter: :equal,
    timeout_ms: 5_000
  )

flaky_graph = put_in(graph.nodes_by_name["location_model"].retry, flaky_retry)
flaky_graph.nodes_by_name["location_model"].retry

backoff and jitter are strategies: a module that implements DomovoyCore.Retry.Backoff or DomovoyCore.Retry.Jitter with its options. The struct above holds that {module, opts} form; the shorthands in the cell are what Retry.new/2 expands. {:exponential, 250, 2, 2_000} waits 250 ms before the first retry and doubles the wait for each later one, up to 2 s; {:fixed, ms} — or the older backoff_ms: ms — waits the same every time. Retry.backoff_ms/3 gives the wait before the retry that follows an attempt, and Retry.delay_ms/3 the same wait after jitter. :equal draws from half the backoff to the whole, :full from zero to the whole, so runs that fail together do not retry together. The default, :none, keeps the exact backoff.

for attempt <- 1..3 do
  %{
    after_attempt: attempt,
    backoff_ms: Retry.backoff_ms(flaky_retry, attempt),
    delay_ms: Retry.delay_ms(flaky_retry, attempt)
  }
end

Backoff strategies side by side

Five built-in backoffs, one row per retry. Every column starts at 250 ms; the difference is how fast each one reaches its 2 s cap. :decorrelated draws each wait from the base up to three times the previous one, so the cell threads the previous wait through Retry.Info the way the engine does inside a run — evaluate it again and that column moves, the others do not.

alias DomovoyCore.Retry.Info

backoffs = [
  fixed: Retry.new(backoff: {:fixed, 250}),
  linear: Retry.new(backoff: {:linear, 250, 250, 2_000}),
  exponential: Retry.new(backoff: {:exponential, 250, 2, 2_000}),
  decorrelated: Retry.new(backoff: {:decorrelated, 250, 2_000}),
  schedule: Retry.new(backoff: {:schedule, [250, 250, 1_000, 2_000]})
]

# One column per strategy, ten retries deep. `previous` carries the wait of
# each strategy into its next retry, as the engine does.
{backoff_rows, _previous} =
  Enum.map_reduce(1..10, %{}, fn retry, previous ->
    waits =
      Map.new(backoffs, fn {name, policy} ->
        info = Info.new(previous_ms: previous[name])
        {name, Retry.backoff_ms(policy, retry, info)}
      end)

    {Map.put(waits, :retry, retry), waits}
  end)

backoff_rows

The same ten retries as a running total — the time a node has spent waiting before retry n starts, with no time for the attempts themselves. This is the number to hold against a deadline_ms: a policy whose total passes the deadline never reaches that retry, whatever max_attempts allows.

totals =
  Enum.scan(backoff_rows, %{}, fn row, sums ->
    Map.new(row, fn
      {:retry, retry} -> {:retry, retry}
      {name, wait} -> {name, Map.get(sums, name, 0) + wait}
    end)
  end)

totals

With deadline_ms: 5_000, read the exponential column: its total passes 5 s at retry 5, so the engine refuses that retry and attempt 5 is the last, even with max_attempts: 10. fixed still has retries to spare. The cell lists, per retry, the strategies whose total still fits the budget.

for row <- totals, into: %{} do
  fits = for {name, total} <- row, name != :retry, total <= 5_000, do: name
  {row.retry, Enum.sort(fits)}
end

Jitter modes: desynchronizing concurrent retries

Jitter spreads the waits so that many concurrent runs do not retry in lockstep. Four built-in strategies on the same exponential backoff, one row per retry: the exact backoff, then one draw of each jitter. Evaluate the cell again and the draws move; the bounds do not.

exponential = {:exponential, 250, 2, 2_000}

jitters = [
  none: Retry.new(backoff: exponential, jitter: :none),
  equal: Retry.new(backoff: exponential, jitter: :equal),
  full: Retry.new(backoff: exponential, jitter: :full),
  proportional_20: Retry.new(backoff: exponential, jitter: {:proportional, 0.2})
]

for retry <- 1..10 do
  jitters
  |> Map.new(fn {name, policy} -> {name, Retry.delay_ms(policy, retry)} end)
  |> Map.put(:retry, retry)
end

• :none (default): exact wait. All 100 concurrent runs retry together.
• :equal: between half the backoff and the full backoff. Spreads retries across a narrower range — good for gentle de-synchronization.
• :full: between zero and the full backoff. Spreads retries across the widest range — good when you want maximum separation of retry storms.
• {:proportional, 0.2}: the backoff plus or minus a fifth. The one that may wait longer than the backoff, so its total against a deadline can be a fifth higher than the :none column above.

Why jitter, and why a decorrelated backoff, in one table. A hundred runs fail at the same moment and retry three times. The cell adds up the waits of each run and asks, for every policy, how many of the hundred come back inside the same 100 ms window at their third retry — the size of the largest herd — and how many windows the hundred spread over. :none sends all hundred together. The jitters spread them, :full the most, but every run on its third retry still draws from the same band of the exponential curve. :decorrelated grows each wait from the previous one instead of the attempt number, so runs that drew differently once diverge further at every later retry, and a run that drew a long wait can draw a short one next: the range always starts at the base. Evaluate the cell a few times.

herd_policies =
  jitters ++ [decorrelated: Retry.new(backoff: {:decorrelated, 250, 2_000})]

# When does run `i` come back for its third retry? The sum of its three waits,
# each grown from the previous one the way the engine does.
third_retry_at = fn policy ->
  for _run <- 1..100 do
    {waits, _previous} =
      Enum.map_reduce(1..3, nil, fn retry, previous ->
        wait = Retry.delay_ms(policy, retry, Info.new(previous_ms: previous))
        {wait, wait}
      end)

    Enum.sum(waits)
  end
end

for {name, policy} <- herd_policies do
  windows = policy |> third_retry_at.() |> Enum.frequencies_by(&div(&1, 100))

  %{
    policy: name,
    largest_herd: Enum.max(Map.values(windows)),
    windows_used: map_size(windows),
    earliest_ms: windows |> Map.keys() |> Enum.min() |> Kernel.*(100),
    latest_ms: windows |> Map.keys() |> Enum.max() |> Kernel.*(100)
  }
end

That is the trade: :full breaks up the herd well and :decorrelated better still, over a wider and less predictable span. Marc Brooker's simulation for the AWS Architecture Blog, where the decorrelated jitter comes from, found the two close in completion time with the decorrelated one making slightly more calls, and both far ahead of :none and :equal. :decorrelated is the one to reach for when many runs contend for one shared thing and a storm costs more than a predictable curve; Polly's recommended DecorrelatedJitterBackoffV2 and redis-py's DecorrelatedJitterBackoff are the same idea. Exponential with :full is the one to keep when a deadline_ms has to be budgeted, because its running total is the :none column of the totals table at most; the decorrelated total is not.

Every strategy draws through the random of its Retry.Info, and Engine.run/4 takes a random_fun: that replaces it for the whole run. The bounds of each jitter are then plain to see: a draw that always gives zero shows the least each one waits, a draw that always gives the maximum the most.

least = Info.new(random: fn _max -> 0 end)
most = Info.new(random: fn max -> max end)

for retry <- 1..10 do
  jitters
  |> Enum.flat_map(fn {name, policy} ->
    [
      {:"#{name}_min", Retry.delay_ms(policy, retry, least)},
      {:"#{name}_max", Retry.delay_ms(policy, retry, most)}
    ]
  end)
  |> Map.new()
  |> Map.put(:retry, retry)
end

A strategy of your own sees the same Retry.Info: the error that failed the attempt, the previous wait, what is left of the deadline and the draw. This one honours a retry_after_ms that a service put in the reason of its error and falls back to the exponential curve otherwise — the case of a rate limit that says when to come back.

defmodule Cracow.Retry.RetryAfter do
  @moduledoc "Waits what the error asked for, or falls back to another backoff."
  @behaviour DomovoyCore.Retry.Backoff

  alias DomovoyCore.Retry.Backoff

  @impl true
  def validate(fallback: fallback), do: Backoff.validate_strategy(fallback)
  def validate(_opts), do: {:error, "takes fallback, a backoff strategy"}

  @impl true
  def backoff_ms([fallback: fallback], attempt, info) do
    case info.error do
      %DomovoyCore.Error{reason: %{retry_after_ms: ms}} when is_integer(ms) and ms >= 0 ->
        ms

      _other ->
        {module, opts} = Backoff.normalize(fallback)
        module.backoff_ms(opts, attempt, info)
    end
  end
end

retry_after = Retry.new(max_attempts: 5, backoff: {Cracow.Retry.RetryAfter, fallback: exponential})

asked = %DomovoyCore.Error{type: :rate_limited, reason: %{retry_after_ms: 1_500}, retryable?: true}
silent = %DomovoyCore.Error{type: :location_unavailable, retryable?: true}

for {label, error} <- [asked: asked, silent: silent], retry <- 1..3 do
  %{error: label, retry: retry, wait_ms: Retry.delay_ms(retry_after, retry, Info.new(error: error))}
end

The runner below fails its first two attempts and computes on the third. It knows the attempt from the job: the engine counts attempts inside one run, so context.job.attempt is 1, then 2, then 3. Only an error with retryable?: true retries — a refusal like the validator's never would.

defmodule Cracow.Runner.FlakyLocation do
  @moduledoc "Fails its first two attempts with a retryable error, then computes."
  use DomovoyCore.Runner

  alias DomovoyCore.Error

  input do
    field :features, DomovoyCore.Type.Map
  end

  required [:features]

  @impl DomovoyCore.Runner
  def run(%Input{features: features}, %DomovoyCore.Context{job: job}) do
    if job.attempt < 3 do
      {:error, Error.new(%{type: :location_unavailable, retryable?: true})}
    else
      %{"district" => district, "floor" => floor, "balcony" => balcony, "year_built" => year} =
        features

      {:ok, Cracow.Market.location_factor(district, floor, balcony, year)}
    end
  end
end

A fresh context with its own files, and the engine called directly with the override — the bullet's shape. The listing is cast through the graph inputs exactly as the first notebook's helper did.

retry_root = Path.join(System.tmp_dir!(), "domovoy_cracow_retry")
File.rm_rf!(retry_root)

open_retry_run = fn run_id ->
  {:ok, store} =
    Store.open(Store.FileSystem, run_id, workflow: workflow.name, root: retry_root)

  {:ok, journal} =
    Journal.open(Cracow.Runtime, Journal.FileSystem, run_id,
      workflow: workflow.name,
      root: retry_root
    )

  %Context{
    job: Job.new(run_id),
    workflow: workflow.name,
    stage: "estimate_price",
    store: store,
    journal: journal,
    runtime: Cracow.Runtime
  }
end

retry_context = open_retry_run.("flaky-1")

typed =
  Map.new(listing, fn {name, raw} ->
    {name, Value.cast!(raw, Map.fetch!(flaky_graph.inputs, name))}
  end)

{:ok, flaky_records} =
  Engine.run(flaky_graph, typed, retry_context,
    runners: %{"location_model" => Cracow.Runner.FlakyLocation}
  )

Cracow.Sale.pln(flaky_records["estimate"].result.value["price"])

Every attempt is its own record, and the store keeps all three: two errors, then the value. The run succeeds because the third attempt does.

{:ok, kept} = Store.all(retry_context.store)

kept
|> Enum.filter(&(&1.node == "location_model"))
|> Enum.sort_by(& &1.job.attempt)
|> Cracow.Sale.records_table()

The journal kept the waits too: one :node_retried per backoff, carrying the job of the attempt that comes next, "backoff_ms", the exact wait of the strategy — 250 ms before attempt 2, 500 ms before attempt 3 — and "wait_ms", what :equal jitter made of it: between 125 and 250 ms, then between 250 and 500 ms. Evaluate the cell above again and the waits move; the backoffs and the bounds do not. The [:node, :retry] telemetry event of the same moment carries more: the two numbers, the previous wait, what was left of the deadline, the error, and the whole policy — everything the engine had in hand.

{:ok, retry_events} = Journal.events(retry_context.journal)

for %DomovoyCore.Event{kind: :node_retried} = event <- retry_events,
    do: %{next_attempt: event.job.attempt, payload: event.payload}

An exhausted retry fails the run

The same graph and the same policy, but a runner that never recovers. Three attempts fail, the budget is spent, and the failure is terminal: the estimate never runs, so it is skipped, and the run returns the error with every record the attempts left behind.

defmodule Cracow.Runner.AlwaysDown do
  @moduledoc "Never recovers: every attempt fails with a retryable error."
  use DomovoyCore.Runner

  alias DomovoyCore.Error

  input do
    field :features, DomovoyCore.Type.Map
  end

  required [:features]

  @impl DomovoyCore.Runner
  def run(%Input{}, %DomovoyCore.Context{}) do
    {:error, Error.new(%{type: :location_unavailable, retryable?: true})}
  end
end

down_context = open_retry_run.("down-1")

{:error, down_error, down_records} =
  Engine.run(flaky_graph, typed, down_context,
    runners: %{"location_model" => Cracow.Runner.AlwaysDown}
  )

{down_error.type, Cracow.Sale.records_table(down_records)}

A deadline caps the budget

max_attempts counts calls; deadline_ms counts time. It is the budget of one node inside one run: it starts with the first attempt of the node and covers every later attempt and the waits between them. The engine refuses an attempt that would start after the deadline — and, sooner, a retry whose wait would end after it — with a :deadline_exceeded error. The refusal is a terminal failure with a record of its own, so the store and the journal show where the budget ran out.

The policy below allows ten attempts but one second in all. Without jitter the waits are exactly 200, 400 and 800 ms: three attempts fail by about 600 ms, and the fourth would start at 1 400 ms, past the deadline. The engine refuses it at once instead of waiting 800 ms for a retry it would refuse anyway, so the run fails in about 600 ms.

deadline_graph =
  put_in(
    graph.nodes_by_name["location_model"].retry,
    Retry.new(max_attempts: 10, backoff: {:exponential, 200, 2, 5_000}, deadline_ms: 1_000)
  )

deadline_context = open_retry_run.("deadline-1")

{elapsed_us, {:error, deadline_error, _deadline_records}} =
  :timer.tc(fn ->
    Engine.run(deadline_graph, typed, deadline_context,
      runners: %{"location_model" => Cracow.Runner.AlwaysDown}
    )
  end)

%{
  error: deadline_error.type,
  reason: deadline_error.reason,
  elapsed_ms: div(elapsed_us, 1_000)
}

Three attempts called the runner; the fourth never did. Its record has no started_at, and in the journal its :node_failed has no :node_started before it — the shape of every attempt the engine refuses before it runs.

{:ok, deadline_kept} = Store.all(deadline_context.store)

deadline_kept
|> Enum.filter(&(&1.node == "location_model"))
|> Enum.sort_by(& &1.job.attempt)
|> Enum.map(&%{attempt: &1.job.attempt, status: &1.status, started_at: &1.started_at})

{:ok, deadline_events} = Journal.events(deadline_context.journal)

for event <- deadline_events, event.subject == "location_model",
    do: {event.kind, event.job.attempt, event.payload}

A breaker stops hammering a runner

A retry policy looks at one node. A breaker looks at the runner behind it: %{threshold: n, cool_down_ms: ms} counts consecutive failed calls of that runner inside one run, across every node that uses it with a breaker, and opens after n. While it is open the engine refuses an attempt of the runner with a :breaker_open error, without calling it, and the run stops. After the cool-down the next attempt probes the runner: success closes the breaker, failure opens it again.

Here the layout model and the location model both run Cracow.Runner.AlwaysDown — the override checks that its input schema matches, and both nodes take one features map — with a breaker that opens at two failures. max_concurrency: 1 runs one node at a time, so the story reads top to bottom: one branch fails once, the other branch fails once, the breaker of the runner opens, and the first retry to come up is refused before it can call the runner. The other retry is cancelled with the run.

breaker_retry =
  Retry.new(
    max_attempts: 5,
    backoff: {:fixed, 0},
    breaker: %{threshold: 2, cool_down_ms: 60_000}
  )

breaker_graph = put_in(graph.nodes_by_name["layout_model"].retry, breaker_retry)
breaker_graph = put_in(breaker_graph.nodes_by_name["location_model"].retry, breaker_retry)

breaker_context = open_retry_run.("breaker-1")

{:error, breaker_error, breaker_records} =
  Engine.run(breaker_graph, typed, breaker_context,
    runners: %{
      "layout_model" => Cracow.Runner.AlwaysDown,
      "location_model" => Cracow.Runner.AlwaysDown
    },
    max_concurrency: 1
  )

{breaker_error.type, breaker_error.reason}

The runner ran twice in the whole run, once per branch. The refused attempt is the record with error: breaker_open; the other branch was cancelled while it waited for its retry, and the estimate was skipped.

Cracow.Sale.records_table(breaker_records)

{:ok, breaker_events} = Journal.events(breaker_context.journal)

for event <- breaker_events, event.subject in ["layout_model", "location_model"],
    do: {event.subject, event.kind, event.job.attempt, event.payload}

Where to go next

• The review here waits on Run.decide/4 in a notebook cell. In a real host a LiveView renders the prompt, the choices and the records, and calls Server.decide/3 on the run's owner — the same calls as above, behind buttons.
• The forecast override is hand-written. A DomovoyCore.Decider that reads the estimate from the store can answer cheap flats on its own and leave only the expensive ones to a person: {:ok, choice} under a threshold, :await above it.
• The engine events of the first notebook never needed the journal, and neither does a dashboard: a Telemetry.Metrics.summary/2 of the duration of [:domovoy_core, :engine, :node, :stop] per node, next to a counter of [:domovoy_core, :engine, :node, :retry] and one of [:domovoy_core, :engine, :node, :breaker_open] per runner, is the latency and the flakiness of every branch in production.