Cracow house prices on the DomovoyCore engine

What this notebook shows

A flat in Kraków gets a price from the number of rooms, the number of bathrooms, the area in square metres and the district, plus a few options: the floor, a balcony, the year the building went up. The model is small and deterministic. The subject of the notebook is the engine that runs it.

The estimate is a diamond-shaped graph run by DomovoyCore.Engine as one DomovoyCore.Stage:

• normalize checks the listing and turns it into one feature map;
• area_model, layout_model and location_model each compute one term of the price. They depend only on normalize, so the Engine runs them in parallel;
• estimate joins the three terms into a price.

Along the way the notebook explains the design choices behind the engine and watches them happen: a custom DomovoyCore.Type, a DomovoyCore.Validator, typed runners, arrows derived from bindings, the store that remembers every record, the journal that remembers every event, the events themselves as they arrive over PubSub and telemetry, what the Engine does when it meets the same generation twice — with a value, or with a failure — and what the Engine measures about itself: the process, the duration and the messages of every attempt.

Every coefficient in this notebook is illustrative. Nothing is fitted to real Kraków listings, so treat the numbers as a demonstration of the engine, not as a valuation.

Start a runtime

DomovoyCore is a library: it starts no global process. The host owns a named DomovoyCore.Runtime — a supervisor with a Phoenix.PubSub, a Registry and two Task.Supervisors — and every engine call names it. In Livebook the notebook is the host.

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

alias DomovoyCore.{Context, Event, Graph, Job, Journal, Node, Stage, Store, Type, Value}
alias DomovoyCore.Engine.Order

A type the engine does not have yet

Every value that crosses an arrow has a type, and a type is a module: an Ecto.Type with a cast/2 that may also read and write metadata. The engine ships Integer, String, Boolean, Map and a few more, but no float. A type is one small module, so the notebook adds its own for the area and for the three model terms.

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

  An integer is promoted to a float; anything else is refused.
  """
  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.Type.Features do
  @moduledoc """
  A `DomovoyCore.Type` for the normalized listing every branch reads.

  A value is a plain map with string keys: `rooms`, `bathrooms`, `area`,
  `district`, `floor`, `balcony`, `year_built` and `m2_per_room`. Anything
  else is refused, so a producer that forgets a field fails at the arrow,
  not inside a branch.
  """
  use DomovoyCore.Type

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

  @impl DomovoyCore.Type
  def cast(raw, _metadata) when is_map(raw) and not is_struct(raw) do
    with %{
           "rooms" => rooms,
           "bathrooms" => bathrooms,
           "area" => area,
           "district" => district,
           "floor" => floor,
           "balcony" => balcony,
           "year_built" => year_built,
           "m2_per_room" => m2_per_room
         } <- raw,
         true <- is_integer(rooms),
         true <- is_integer(bathrooms),
         true <- is_number(area),
         true <- is_binary(district),
         true <- is_integer(floor),
         true <- is_boolean(balcony),
         true <- is_integer(year_built),
         true <- is_number(m2_per_room) do
      {:ok, raw}
    else
      _ -> :error
    end
  end

  def cast(_raw, _metadata), do: :error
end

DomovoyCore.Value.cast/2 is the one constructor of a typed value. It refuses what the type refuses, and the error names the type but never the value — that rule holds for every error the engine makes, so a log of a failed run never leaks what was in the data.

{Value.cast(58.5, Cracow.Type.Float), Value.cast(60, Cracow.Type.Float),
 Value.cast("58.5", Cracow.Type.Float)}

The price model

Three terms, multiplied together. The base value is a quadratic in the area (small flats cost more per metre), the layout factor rewards rooms and bathrooms but penalises too many rooms for the area, and the location factor is a district multiplier times a few small steps.

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

  Every coefficient is made up to look plausible; nothing is fitted to real
  listings.

      price = base(area) × layout(rooms, bathrooms, area) × location(district, floor, balcony, year)
  """

  @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)

  @doc "PLN per m² as a quadratic in area: small flats cost more per metre."
  def base_per_m2(area), do: 17_800 - 42 * area + 0.16 * area * area

  @doc "Value of the flat before layout and location adjustments."
  def base_value(area), do: area * base_per_m2(area)

  @doc "Rooms and bathrooms add value; too many rooms for the area take it away."
  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

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

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

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

{Cracow.Market.base_value(58.5), Cracow.Market.layout_factor(3, 1, 58.5),
 Cracow.Market.location_factor("Krowodrza", 3, false, 2010)}

Runners: the four corners of the diamond

A runner is the unit of work. It declares its input as an Ecto schema, so the contract of a node is data the engine can check, not a convention. The Engine casts the parameters, runs the validators on the changeset, and only then calls run/2. A runner that gets called can trust its input; a runner that would be called with bad input is never called at all.

Why split a small formula into five runners? Because the engine schedules runners, stores their records, journals their events and retries them one by one. The graph is the unit of explanation: the store will show which term came from where, and the branches that do not depend on each other run at the same time.

defmodule Cracow.Validator.Listing do
  @moduledoc """
  Refuses a listing no flat in Kraków could match. Errors name fields and
  rules, never values.
  """
  @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_or_equal_to: 15)
    |> validate_number(:floor, greater_than_or_equal_to: 0)
    |> validate_inclusion(:district, Cracow.Market.districts())
    |> validate_bathrooms_within_rooms()
  end

  defp validate_bathrooms_within_rooms(changeset) do
    rooms = get_field(changeset, :rooms)
    bathrooms = get_field(changeset, :bathrooms)

    if is_integer(rooms) and is_integer(bathrooms) and bathrooms > rooms do
      add_error(changeset, :bathrooms, "must not exceed rooms",
        validation: :bathrooms_within_rooms
      )
    else
      changeset
    end
  end
end

The top of the diamond. Its input fields are the graph inputs; its output is the one Cracow.Type.Features value every branch reads. The value is a plain map with string keys because a record is written to disk as JSON and read back, and a value must survive that round trip unchanged.

defmodule Cracow.Runner.Normalize do
  @moduledoc "Checks the listing and gives one features value 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

The three branches. Each binds the same features value and gives one float.

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

  input do
    field :features, Cracow.Type.Features
  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, Cracow.Type.Features
  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, Cracow.Type.Features
  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

The bottom of the diamond. It binds all three branches, so the Engine starts it only when the last of them has finished. It also binds normalize for the area, which shows that a binding may skip a level: an arrow is any producer to any consumer, not only neighbour to neighbour.

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

  input do
    field :features, Cracow.Type.Features
    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 and its stage

A node names its runner, its output type and its bindings. Nobody draws an edge: Graph.new/1 derives every arrow from the bindings, and a source that no node produces becomes a graph input. Every binding type must equal both the runner's field type and the producer's output type, so a mismatch fails here, when the graph is built, not later inside a run.

graph =
  Graph.new([
    Node.new(%{
      name: "normalize",
      runner: Cracow.Runner.Normalize,
      type: Cracow.Type.Features,
      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", Cracow.Type.Features}}
    }),
    Node.new(%{
      name: "layout_model",
      runner: Cracow.Runner.LayoutModel,
      type: Cracow.Type.Float,
      bind: %{features: {"normalize", Cracow.Type.Features}}
    }),
    Node.new(%{
      name: "location_model",
      runner: Cracow.Runner.LocationModel,
      type: Cracow.Type.Float,
      bind: %{features: {"normalize", Cracow.Type.Features}}
    }),
    Node.new(%{
      name: "estimate",
      runner: Cracow.Runner.Estimate,
      type: Type.Map,
      bind: %{
        features: {"normalize", Cracow.Type.Features},
        base: {"area_model", Cracow.Type.Float},
        layout: {"layout_model", Cracow.Type.Float},
        location: {"location_model", Cracow.Type.Float}
      }
    })
  ])

stage = Stage.new(%{name: "estimate_price", graph: graph})

A stage is a named wrapper around one graph and a vertex of a workflow. It declares no output of its own: the value of every node sits in the store under the name of the node, and a node of a later stage binds to it by name. The stage knows what it needs from outside, and Order.waves/1 shows which nodes may run together. The middle wave is the parallel one.

{Stage.inputs(stage), Order.waves(graph)}

The diagram below is what the Engine sees. Livebook renders the static mermaid block natively, without Kino. The Elixir cell after it regenerates the same source from graph.arrows and graph.inputs, so the picture and the engine cannot drift apart unnoticed.

flowchart LR
  subgraph inputs
    area([area]):::input
    balcony([balcony]):::input
    bathrooms([bathrooms]):::input
    district([district]):::input
    floor([floor]):::input
    rooms([rooms]):::input
    year_built([year_built]):::input
  end
  area_model[area_model]
  estimate[estimate]
  layout_model[layout_model]
  location_model[location_model]
  normalize[normalize]
  area --> normalize
  area_model --> estimate
  balcony --> normalize
  bathrooms --> normalize
  district --> normalize
  floor --> normalize
  layout_model --> estimate
  location_model --> estimate
  normalize --> area_model
  normalize --> estimate
  normalize --> layout_model
  normalize --> location_model
  rooms --> normalize
  year_built --> normalize
  classDef input fill:none,stroke-dasharray:4 3

inputs = graph.inputs |> Map.keys() |> Enum.sort()
nodes = graph.nodes_by_name |> Map.keys() |> Enum.sort()

mermaid =
  ["flowchart LR", "  subgraph inputs"] ++
    Enum.map(inputs, &"    #{&1}([#{&1}]):::input") ++
    ["  end"] ++
    Enum.map(nodes, &"  #{&1}[#{&1}]") ++
    Enum.map(graph.arrows, &"  #{&1.from} --> #{&1.to}") ++
    ["  classDef input fill:none,stroke-dasharray:4 3"]

IO.puts(Enum.join(mermaid, "\n"))

The store and the journal

A run has two memories, and the engine writes both through adapters.

The store holds one record per node attempt, under the key {run_id, node, generation, attempt}. A record is the outcome of one attempt: its status, its typed value or its error, and when it started and finished. The store is how a later stage reads an earlier one, how a resumed run knows what it already did, and how a person inspects a run afterwards.

The journal is an append-only list of events. Nothing in it changes or goes away. The state of a run is what its events say, so a run can be rebuilt by folding them, and every append is also broadcast on the runtime's PubSub and executed as telemetry, so anything can watch a run live.

Both adapters here write files, so the run survives the notebook. The root is cleared every time this cell runs, so the sections below start from an empty run.

root = Path.join(System.tmp_dir!(), "domovoy_cracow")
File.rm_rf!(root)
workflow = "cracow_prices"

open = fn %Job{} = job ->
  {:ok, store} = Store.open(Store.FileSystem, job.id, workflow: workflow, root: root)

  {:ok, journal} =
    Journal.open(Cracow.Runtime, Journal.FileSystem, job.id, workflow: workflow, root: root)

  %Context{
    job: job,
    workflow: workflow,
    stage: stage.name,
    store: store,
    journal: journal,
    runtime: Cracow.Runtime
  }
end

context = open.(Job.new("flat-1"))

A few helpers for the cells that follow: typed inputs from a raw listing, a listener that collects what the run broadcasts, and tables.

defmodule Cracow.Notebook do
  @moduledoc "Glue between the notebook cells and the engine."

  alias DomovoyCore.{Context, Engine, Event, Journal, Runtime, Stage, Value}

  @doc "Casts a raw listing into typed graph inputs. The stage says which type each input has."
  def inputs(%Stage{} = stage, listing) do
    types = Stage.inputs(stage)
    Map.new(listing, fn {name, raw} -> {name, Value.cast!(raw, Map.fetch!(types, name))} end)
  end

  @doc """
  Runs the stage while listening to its events, and gives the engine result
  with everything that arrived over PubSub and telemetry, in order.
  """
  def run(%Stage{} = stage, listing, %Context{} = context, opts \\ []) do
    topic = Journal.topic(context.workflow, context.job.id)
    :ok = Phoenix.PubSub.subscribe(Runtime.pubsub(context.runtime), topic)
    names = Event.telemetry_events()
    :ok = :telemetry.attach_many(topic, names, &__MODULE__.forward_telemetry/4, self())
    _stale = drain([])

    try do
      result = Engine.run(stage.graph, inputs(stage, listing), context, opts)
      {result, drain([])}
    after
      :telemetry.detach(topic)
      Phoenix.PubSub.unsubscribe(Runtime.pubsub(context.runtime), topic)
    end
  end

  @doc false
  def forward_telemetry(name, _measurements, metadata, pid),
    do: send(pid, {:telemetry, name, metadata})

  defp drain(acc) do
    receive do
      {:domovoy_event, %Event{} = event} -> drain([{:pubsub, event} | acc])
      {:telemetry, name, metadata} -> drain([{:telemetry, name, metadata} | acc])
    after
      0 -> Enum.reverse(acc)
    end
  end

  @doc "The value of the `estimate` node."
  def estimate(records), do: records["estimate"].result.value

  @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 "One row per event."
  def events_table(events) do
    for %Event{} = event <- events do
      %{
        at: Calendar.strftime(event.at, "%H:%M:%S.%f"),
        kind: event.kind,
        subject: event.subject,
        generation: event.job.generation,
        attempt: event.job.attempt,
        payload: inspect(event.payload)
      }
    end
  end

  @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 the stage and listen

The listing is a plain map. The engine never sees it as such: inputs/2 casts each entry through the type the stage declares for it, and the Engine gets typed values.

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

{{:ok, records}, live} = Cracow.Notebook.run(stage, listing, context)
estimate = Cracow.Notebook.estimate(records)

IO.puts("""
## #{Cracow.Notebook.pln(estimate["price"])}

**#{Cracow.Notebook.pln(estimate["per_m2"])} / m²** ·
base #{Cracow.Notebook.pln(estimate["components"]["base"])}
× layout #{Float.round(estimate["components"]["layout"], 3)}
× location #{Float.round(estimate["components"]["location"], 3)}
""")

estimate

The result is one record per graph node, and the value of estimate is the output of the stage. The graph inputs are not in the result — they were given, not computed — but they are in the store, as the next section shows.

Cracow.Notebook.records_table(records)

Every event arrived twice while the stage ran: once as {:domovoy_event, event} over PubSub on the topic run:<workflow>:<run_id>, and once as telemetry under [:domovoy_core, :event, <kind>]. PubSub is for processes that follow one run — a LiveView, a notebook. Telemetry is for the metrics and logging of the whole system. The order is the order of the journal, because the journal commits the append before it broadcasts.

pubsub = for {:pubsub, event} <- live, do: event
telemetry = for {:telemetry, name, _metadata} <- live, do: name

IO.puts("#{length(pubsub)} events over PubSub, #{length(telemetry)} telemetry executions")

Cracow.Notebook.events_table(pubsub)

Read the events top to bottom. node_started for the three branches comes before any of their node_finished, because each runner runs as its own task under the runtime's Task.Supervisor, and the Engine starts a node the moment its last predecessor finishes. max_concurrency caps the tasks; Order.waves/1 is the shape of the graph, not the schedule.

Look inside the store and the journal

The store holds the seven inputs and the five nodes at generation 0, attempt 1. The value of each node is here, typed. A later stage that binds "estimate" reads this record, so a stage never needs to hand its values onward — the store is the interface between stages.

{:ok, stored} = Store.all(context.store)
Cracow.Notebook.records_table(stored)

The journal holds the same events that arrived live, now on disk.

{:ok, journaled} = Journal.events(context.journal)
Cracow.Notebook.events_table(journaled)

Both are files. One JSON document per record, named after the node, the generation and the attempt, and one journal file with one event per line. A person with a shell and jq can read a run without the engine.

run_directory = Path.join([root, workflow, context.job.id])

files =
  run_directory
  |> File.ls!()
  |> Enum.sort()
  |> Enum.map(&%{file: &1, bytes: File.stat!(Path.join(run_directory, &1)).size})

IO.inspect(files, label: run_directory)

File.read!(Path.join(run_directory, "estimate-g0-a1.json"))

Run it again: the store answers instead of the runners

Before it schedules anything, the Engine asks the store for the record of every node at the exact generation of the job. A record there is a hit: the node does not run again, its stored record stands for it, and its successors read that record as if it had just been produced. Run the same job once more and no runner runs. The events say so: every node_finished carries "hit" => true, and there is no node_started at all.

{{:ok, again}, live_again} = Cracow.Notebook.run(stage, listing, context)

same? = Cracow.Notebook.estimate(again) == estimate
IO.puts("same value as the first run: #{same?}")

Cracow.Notebook.events_table(for {:pubsub, event} <- live_again, do: event)

This is what makes a run resumable. A process that dies halfway leaves its finished nodes in the store; the next process runs the same job and only does the work that is missing.

A failed generation replays as a failure

A listing with zero rooms fails the validator. The Engine never calls the runner, the result carries the error, and every node still gets a final record: normalize is :error, and the four nodes that never started are :skipped. All five records go to the store.

failed_context = open.(Job.new("flat-2"))
{{:error, error, failed}, _live} =
  Cracow.Notebook.run(stage, %{listing | "rooms" => 0}, failed_context)

IO.puts("#{error.type} — #{inspect(error.reason)}")
Cracow.Notebook.records_table(failed)

Now run the same job again. The store holds a record for every node at generation 0, and a record of any status is a hit. The Engine halts before it runs anything, keeps the stored records, and fails with the stored error. The events are node_failed and node_skipped, each with "hit" => true, and there is no node_started. The error and the records come back as the store kept them — read from their JSON documents — so the reason of the error is in its document form, with string keys where the original had a keyword list.

{{:error, replayed_error, replayed}, live_replay} =
  Cracow.Notebook.run(stage, %{listing | "rooms" => 0}, failed_context)

statuses = fn records -> Map.new(records, fn {name, record} -> {name, record.status} end) end
started = for {:pubsub, %Event{kind: :node_started}} <- live_replay, do: 1

IO.puts("""
same error type as the first run: #{replayed_error.type == error.type} ·
same statuses as the first run: #{statuses.(replayed) == statuses.(failed)} ·
nodes started: #{length(started)}
""")

Cracow.Notebook.events_table(for {:pubsub, event} <- live_replay, do: event)

The reasoning: a generation is a fact about the past, and the store is its memory. If the Engine ran the failed node again at the same generation, a replay could give a different answer from the run it replays, and a resumed run could not trust what it reads. So the Engine never runs a generation twice, for a value or for a failure. A retry is a new generation. That is what a {:rerun, stage} target of a DomovoyCore.Decision does in a workflow, and what Job.next_generation/1 does here. The old records stay; the new ones sit next to them.

fixed_context = %Context{failed_context | job: Job.next_generation(failed_context.job)}
{{:ok, fixed}, _live} = Cracow.Notebook.run(stage, listing, fixed_context)

fixed_price = Cracow.Notebook.pln(Cracow.Notebook.estimate(fixed)["price"])
IO.puts("generation 1: #{fixed_price}")

{:ok, both_generations} = Store.all(fixed_context.store)
Cracow.Notebook.records_table(both_generations)

Store.get/3 — the exact-generation read the Engine uses for hits — gives the failed record at generation 0 and the value at generation 1. Store.latest/3 — the read a later stage uses for a binding — walks back through generations but only ever gives a value, because a binding needs one.

{Store.get(fixed_context.store, "normalize", 0),
 Store.get(fixed_context.store, "normalize", 1),
 Store.latest(fixed_context.store, "estimate", 1)}

The Engine's own telemetry

The telemetry so far mirrored the journal: one execution per event under [:domovoy_core, :event, <kind>], and only when the run has a journal. It says what happened to a run. The Engine executes a second family from the coordinator process, under [:domovoy_core, :engine, ...], with or without a journal, and it says how: which process ran which attempt, how long it took, and which message moved the coordinator. DomovoyCore.Telemetry is the catalogue of the names, and DomovoyCore.Engine.Telemetry builds the metadata. Every map holds workflow, run_id, generation and stage, so a handler tells runs apart without state of its own.

A point that marks a moment measures system_time and monotonic_time; a point that closes something measures duration and monotonic_time, in the native unit of System.monotonic_time/0. The coordinator is the process that calls Engine.run/4 — here, the notebook — so a handler that sends to self() collects the trace in its own mailbox, in the order of execution.

defmodule Cracow.Trace do
  @moduledoc """
  Collects the engine telemetry of one `Engine.run/4`, in the order it was
  executed, and turns it into tables.
  """

  alias DomovoyCore.{Context, Engine, Stage}

  @suffixes [
    [:run, :start],
    [:run, :stop],
    [:run, :exception],
    [:node, :hit],
    [:node, :start],
    [:node, :stop],
    [:node, :retry],
    [:node, :breaker_open],
    [:node, :cancelled],
    [:node, :skipped],
    [:task, :start],
    [:task, :stop],
    [:message]
  ]

  @doc "Every engine event, named through the catalogue."
  def events, do: Enum.map(@suffixes, &DomovoyCore.Telemetry.event(:engine, &1))

  @doc """
  Runs the stage with a handler on every engine event, and gives the engine
  result with the trace: one `{suffix, measurements, metadata}` per execution.
  """
  def run(%Stage{} = stage, listing, %Context{} = context, opts \\ []) do
    handler = {__MODULE__, context.job.id, context.job.generation}
    :ok = :telemetry.attach_many(handler, events(), &__MODULE__.forward/4, self())
    _stale = drain([])

    try do
      result = Engine.run(stage.graph, Cracow.Notebook.inputs(stage, listing), context, opts)
      {result, drain([])}
    after
      :telemetry.detach(handler)
    end
  end

  @doc false
  def forward([:domovoy_core, :engine | suffix], measurements, metadata, pid),
    do: send(pid, {:engine_telemetry, suffix, measurements, metadata})

  defp drain(acc) do
    receive do
      {:engine_telemetry, suffix, measurements, metadata} ->
        drain([{suffix, measurements, metadata} | acc])
    after
      0 -> Enum.reverse(acc)
    end
  end

  @doc "The duration of the `run.stop` event, in microseconds."
  def run_us(trace) do
    {_suffix, %{duration: duration}, _meta} = Enum.find(trace, &match?({[:run, :stop], _, _}, &1))
    us(duration)
  end

  @doc "One row per execution: the event, its node, its pid, its duration and what it says."
  def table(trace) do
    for {suffix, measurements, meta} <- trace do
      %{
        event: Enum.join(suffix, "."),
        node: meta[:node] || "",
        pid: if(meta[:pid], do: inspect(meta.pid), else: ""),
        duration_us: if(measurements[:duration], do: us(measurements.duration)),
        detail: describe(suffix, meta)
      }
    end
  end

  @doc "One row per attempt, with every time in microseconds since `run.start`."
  def timeline(trace) do
    {_suffix, %{monotonic_time: t0}, _meta} =
      Enum.find(trace, &match?({[:run, :start], _, _}, &1))

    since_start = fn time -> us(time - t0) end

    starts =
      for {[:node, :start], %{monotonic_time: t}, %{node: node, attempt: attempt}} <- trace,
          into: %{},
          do: {{node, attempt}, since_start.(t)}

    tasks =
      for {[:task, :stop], %{duration: d}, %{node: node, attempt: attempt}} <- trace,
          into: %{},
          do: {{node, attempt}, us(d)}

    rows =
      for {[:node, :stop], %{duration: d, monotonic_time: t}, %{node: node, attempt: attempt}} <-
            trace do
        %{
          node: node,
          attempt: attempt,
          started_at_us: starts[{node, attempt}],
          finished_at_us: since_start.(t),
          node_us: us(d),
          task_us: tasks[{node, attempt}]
        }
      end

    Enum.sort_by(rows, & &1.started_at_us)
  end

  defp describe([:run, :start], meta),
    do: "nodes=#{meta.nodes} max_concurrency=#{meta.max_concurrency}"

  defp describe([:run, :stop], meta),
    do: "outcome=#{meta.outcome}#{error(meta)} nodes=#{inspect(meta.nodes_by_status)}"

  defp describe([:node, :start], meta),
    do:
      "attempt=#{meta.attempt} ordinal=#{meta.ordinal} runner=#{inspect(meta.runner)} " <>
        "after=#{inspect(meta.predecessors)} remaining_ms=#{meta.remaining_ms}"

  defp describe([:node, :stop], meta),
    do: "outcome=#{meta.outcome}#{error(meta)} value_type=#{inspect(meta.value_type)}"

  defp describe([:node, :retry], meta),
    do:
      "backoff_ms=#{meta.backoff_ms} wait_ms=#{meta.wait_ms} " <>
        "next_attempt=#{meta.next_attempt}/#{meta.max_attempts} " <>
        "remaining_ms=#{meta.remaining_ms} error=#{meta.error_type} " <>
        "backoff=#{inspect(meta.backoff_strategy)} jitter=#{inspect(meta.jitter_strategy)}"

  defp describe([:node, :breaker_open], meta),
    do: "runner=#{inspect(meta.runner)} failures=#{meta.failures}/#{meta.threshold}"

  defp describe([:node, _hit_or_classified], meta),
    do: "status=#{meta.status} attempt=#{meta.attempt} cause=#{meta.cause_type}"

  defp describe([:task, :start], meta),
    do: "supervisor=#{inspect(meta.supervisor)} timeout_ms=#{meta.timeout_ms}"

  defp describe([:task, :stop], meta), do: "reason=#{meta.reason}"

  defp describe([:message], meta),
    do: "kind=#{meta.kind} message=#{inspect(meta.message, limit: 4)}"

  defp describe(_suffix, meta), do: inspect(meta, limit: 6)

  defp error(%{error_type: nil}), do: ""
  defp error(%{error_type: type}), do: " error_type=#{type}"

  defp us(native), do: System.convert_time_unit(native, :native, :microsecond)
end

A fresh job, so every runner runs. One row per execution, top to bottom, as the coordinator lived it.

traced_context = open.(Job.new("flat-3"))
{{:ok, _traced}, trace} = Cracow.Trace.run(stage, listing, traced_context)

IO.puts("#{length(trace)} executions · run.stop measured #{Cracow.Trace.run_us(trace)} µs")

Cracow.Trace.table(trace)

node.start for normalize is followed by task.start with a pid under the runtime's engine task supervisor. The message with kind=task_result is that task's result arriving in the coordinator's mailbox; task.stop says the coordinator saw the process go down with reason=result, and node.stop closes the attempt with its duration. Then three node.starts in a row before any of their node.stops: the branches are in flight at the same time.

One message per task is all the coordinator receives. A task links itself to the coordinator so that it dies with it, and the coordinator traps exits for the length of the run; once it has the task's result it unlinks the task and flushes its exit signal, so nothing of the run is left in the mailbox of the process that called Engine.run/4. The timeline below makes the overlap of the branches measurable.

Cracow.Trace.timeline(trace)

task_us runs from the spawn of the runner process to the coordinator taking its result. node_us is the attempt as the coordinator saw it, from node.start to the final record, so it also holds the cast, the validators, the store write and the journal appends. The gap between them is the price of the engine, measured rather than guessed. The gap before normalize starts is the Engine checking the graph, writing the seven inputs to the store and asking it for a record of every node — the same work that makes up the whole of the replay trace below.

The replay from the store is a different trace. There is no node.start, no task.start and no message: five node.hits, and the time of the run is the time of five record reads.

{{:ok, _hits}, hit_trace} = Cracow.Trace.run(stage, listing, traced_context)
attempts = for {[:node, :start], _measurements, _meta} <- hit_trace, do: 1

IO.puts("""
fresh run #{Cracow.Trace.run_us(trace)} µs ·
replay #{Cracow.Trace.run_us(hit_trace)} µs ·
attempts started: #{length(attempts)}
""")

Cracow.Trace.table(hit_trace)

And a failure, at the next generation of the same job. normalize starts, but the validator refuses the listing before a task exists, so its node.stop has outcome=error and there is no pid anywhere in the trace. The four nodes that never started get node.skipped, and run.stop counts them all: statuses is what a dashboard would plot.

failing_context = %Context{traced_context | job: Job.next_generation(traced_context.job)}

{{:error, _error, _records}, failed_trace} =
  Cracow.Trace.run(stage, %{listing | "rooms" => 0}, failing_context)

Cracow.Trace.table(failed_trace)

None of this needed the journal. A bare Engine.run/4 with no store and no journal executes the same events, which makes them the hook for the metrics of a whole system: a Telemetry.Metrics.summary/2 of the duration of [:domovoy_core, :engine, :node, :stop], tagged by workflow, stage and node with unit: {:native, :millisecond}, is a latency distribution per node, and a counter of node.hit next to one of node.start is the hit rate of the store.

Where to go next

• Put the stage in a DomovoyCore.Workflow after a DomovoyCore.Decision, so a person sees 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.
• Pass runners: %{"location_model" => AnotherModule} to Engine.run/4 to swap one branch without touching the graph. The override must declare the same input schema, and the Engine checks that before it starts.
• Give a node retry: with backoff, jitter, deadline and breaker options. max_attempts counts calls; backoff: {:exponential, base, factor, cap} grows delays; jitter: :full spreads concurrent retries; deadline_ms: t sets a time budget across all attempts and waits; breaker: %{threshold: n, cool_down_ms: ms} opens after n failures to protect against cascading failures. Both backoff and jitter are behaviours, so a {MyApp.Strategy, opts} of your own that reads the error of the attempt slots in next to the built-ins. The workflow notebook shows each feature in action.