reactive_dag

Build incremental data pipelines out of Ash resources.

You declare what each table is derived from; the engine works out what to recompute when something upstream changes, and recomputes only that. A resource is a node, its reactive do … end block is the computation, and its rows are the result — there is no separate store, no shadow table, and nothing to keep in sync.

reactive do
  recompute_by :category, to: :expenses, from: :category
  reduce group_by: :category, into: [sum: [amount: :total], count: :n]
end

Edit one expense and one category's total is recomputed. Not the table, not the graph — the row that moved.

Extracted from two apps that independently grew the same engine (the Red Hook cascade pipeline and the u2i compliance portal's model_eval), and now shared by both.

Guides:Getting started · Authoring nodes · LLM nodes · Sources and scanning · Configuration · The seams

A pipeline, end to end

Three resources: raw expenses, a rollup, and a verdict over the rollup. This is the whole thing — there is no wiring file, no registry, and no dispatch to write.

1. The leaf — where data enters

defmodule MyApp.Expenses do
  use Ash.Resource, data_layer: AshPostgres.DataLayer, extensions: [ReactiveDag.Node]

  attributes do
    attribute :id, :string, primary_key?: true
    attribute :category, :string
    attribute :amount, :float
  end

  actions do
    defaults [:read, :destroy]
    create :upsert do upsert?(true); accept([:id, :category, :amount]) end
  end

  reactive do
    leaf? true
    dirties_on [:create, :update, :destroy]   # ordinary writes start the cascade
  end
end

dirties_on is the important line. With it, Ash.create!/1 on this resource marks its key dirty inside the write's own transaction — so a rolled-back write leaves nothing to recompute, and a committed one always leaves a mark. No call site has to remember.

2. The rollup — derived from the leaf

defmodule MyApp.CategoryTotals do
  use Ash.Resource, data_layer: AshPostgres.DataLayer, extensions: [ReactiveDag.Node]

  attributes do
    attribute :category, :string, primary_key?: true
    attribute :total, :float
    attribute :n, :integer
  end

  actions do
    defaults [:read, :destroy]
    create :upsert do upsert?(true); accept([:category, :total, :n]) end
  end

  reactive do
    # "a change to one expense invalidates one category" — this single line
    # supplies the input edge, the grouping, and the read scope.
    recompute_by :category, to: :expenses, from: :category

    reduce group_by: :category, into: [sum: [amount: :total], count: :n]
  end
end

No read:, no key:, no upsert:. The library reads expenses (scoped to the dirty categories), folds each group, and upserts the row into this resource by its own identity.

3. The verdict — derived from the rollup

defmodule MyApp.BudgetHealth do
  use Ash.Resource, data_layer: AshPostgres.DataLayer, extensions: [ReactiveDag.Node]

  attributes do
    attribute :key, :string, primary_key?: true
    attribute :status, :string
    attribute :headroom, :float      # why a table is worth having
  end

  actions do
    defaults [:read, :destroy]
    create :upsert do upsert?(true); accept([:key, :status, :headroom]) end
  end

  reactive do
    reduce over: :category_totals,
           group_by: :category,
           into: fn _cat, [row | _] ->
             %{status: if(row.total < 1000.0, do: "present", else: "failing"),
               headroom: 1000.0 - row.total}
           end
  end
end

A verdict is an ordinary row with a :status column. "What is failing?" is filter(status == "failing") — a plain Ash read, with policies, loads and joins.

4. Run it

plan = ReactiveDag.Node.graph([MyApp.Expenses, MyApp.CategoryTotals, MyApp.BudgetHealth])

{:ok, report} =
  ReactiveDag.Drain.run(plan,
    recompute: ReactiveDag.Node.Recompute,
    key_rule:  ReactiveDag.Node.KeyRule)

graph/2 assembles and validates at that point: every edge resolves, ids are unique, the graph is acyclic, declared attributes exist. An authoring mistake fails here rather than at 3am mid-drain.

Writing two expenses and draining once gives this — depth order, and only the keys that actually moved:

expenses         changed: ["e1", "e2"]           triggered_by: nil
category_totals  changed: ["meals", "travel"]    triggered_by: "expenses"
budget_health    changed: ["meals", "travel"]    triggered_by: "category_totals"

Now change one expense in travel and drain again: category_totals claims ["travel"] only, meals is never read, and budget_health recomputes one row. That proportionality is the whole point of the library.

The %Drain.Report{} is the processing trace — one step per recompute with cell, claimed, changed, triggered_by, duration_us, plus run totals. triggered_by reconstructs the causal chain above.

Configuration

config :reactive_dag,
  repo: MyApp.Repo,          # REQUIRED
  dirty_table: "my_dirty"    # optional; defaults to reactive_dag_dirty

One table, the dirty frontier, and ReactiveDag.Migration.up/1 creates it. Everything else is your own resources with their own migrations. Call ReactiveDag.Config.validate!/0 at boot to catch a misconfiguration there rather than at the first query.

Declaring the computation

Authoring is Ash-first: start with what Ash expresses declaratively and step outward only as far as the shape demands. Every rung writes its rows and reports only the changed keys.

rung	when
`aggregate`	the datastore can do it — one GROUP BY, no rows in the BEAM
`reduce`	a fold over a group
`join`	correlate two sides of one input
`union`	roll many nodes' rows into one queryable table
`per_key`	one call per row — an LLM, an embedding, a fetch
`run :action`	a generic Ash action on this resource
`compute Mod`	arbitrary Elixir

aggregate — the datastore does it: group and aggregate a relationship in ONE query, nothing crossing into the BEAM. Only for relationship aggregates, since Ash has no arbitrary GROUP BY … → rows.

aggregate over: :readings, avg: [flow: :avg_flow], count: :day_count

reduce — an in-BEAM fold. The library reads the input's resource auto-scoped to the dirty keys; group_by: names attributes, into: declares the fold. Keys derive as "gf|2025" from a composite group.

reduce group_by: [:fund, :fy], into: [sum: [amount: :total], count: :n]

Escapes, each independent: query: shapes the read without leaving Ash (fn q, dirty -> … end); a fn group_by/key/into for computed shapes; expand: when one group produces many rows.

join — a left join over ONE input. Sides are attributes (left: :acct) or [key: :acct, where: [kind: "budget"]] discriminator splits. An absent side yields nils, because the declared-vs-observed gap is usually information.

join over: :entries,
     left:  [key: :acct, where: [kind: "budget"]],
     right: [key: :acct, where: [kind: "actual"]],
     outer: true,
     into:  [left: [amount: :budget], right: [amount: :actual]]

union — one row per (input, key) across several inputs, which is how "what is failing anywhere?" becomes one indexed table instead of a scan per cell. Maintained incrementally: a verdict flips, one row updates.

union from: [:category_health, :fund_balance],
      into: [check: :cell, subject: :key, status: :status]

per_key — one action call per input row, with fingerprint: to skip the call when nothing it depends on moved. That skip is the point when the call is an LLM.

per_key :summarise,
  args: [text: :body],
  fingerprint: [:body],
  into: [summary: :summary]

run :action — the Ash-native escape hatch: a generic action on this resource takes (keys, cell_id), does its own writes, and returns the changed keys. Arguments, policies and Ash.run_action testability all still apply.

compute Mod — arbitrary Elixir implementing ReactiveDag.Op. Mirrors Ash's calculate :x, :type, MyModule: the arbitrary case is an entity too, not a schema key beside the declarative ones.

`recompute_by` — the declaration the engine cares about

What unit does a change invalidate? It supplies the input edge, the grouping, the claim resolution and the read scope in one line, which is why it replaces key_rule on combinator nodes.

form	meaning
omitted	key-for-key
`recompute_by :category, to: :expenses, from: :category`	per unit, resolved by lookup
`recompute_by :month, from_key: true`	per unit, parsed from the key's segments
`recompute_by :cell`	redo everything

It is the recompute unit, not the output's grain — percentiles recompute_by :day while their rows are keyed day+percentile.

Input edges: `ref` vs `context`

ref :x (also depends_on [:x], or a combinator's over:) — a recompute edge: when x changes, this node is dirtied.
context :x — a context edge: the node reads x as settled context but is not recomputed when x changes. Still a real input — validated, and ordered by depth so x settles first — it just doesn't propagate.

context earns its keep when recompute is expensive or non-deterministic and consults mutable context it shouldn't be re-triggered by:

reactive do
  compute MyApp.EnhanceMinutes   # an LLM pass
  ref :transcripts               # a transcript change RE-RUNS the LLM
  context :people                # a people edit does NOT — the LLM reads
                                 # current people next time it runs anyway
end

Getting data in

Ordinary writes — dirties_on [:create, :update, :destroy], as in the pipeline above. The mark happens inside the write's own transaction, so no call site has to remember.

A scanner — when the data comes from outside (a crawler, a fleet API, a vendor export), the fetch is effectful and fallible, so it stays outside the drain. That split is a design invariant, not an accident: one unreachable vendor cannot wedge the recompute of everything else, and the drain stays re-runnable.

Writing a scanner

The leaf is an ordinary resource that names its source:

defmodule MyApp.Docs do
  use Ash.Resource, data_layer: AshPostgres.DataLayer, extensions: [ReactiveDag.Node]

  attributes do
    attribute :url, :string, primary_key?: true
    attribute :body, :string
    attribute :content_md5, :string
    attribute :last_seen_at, :utc_datetime_usec   # moves on EVERY poll
  end

  actions do
    defaults [:read, :destroy]
    create :upsert do upsert?(true); accept([:url, :body, :content_md5, :last_seen_at]) end
  end

  reactive do
    leaf? true
    poll MyApp.DocCrawler
    fingerprint [:content_md5]     # what counts as a changed observation
  end
end

The source implements two callbacks. poll/1 fetches, hands what it observed to reconcile/3, and returns the keys that changed:

defmodule MyApp.DocCrawler do
  @behaviour ReactiveDag.Source

  @impl true
  def id, do: :doc_crawler

  @impl true
  def origin, do: %{label: "City site · agendas", url: "https://example.gov"}

  @impl true
  def poll(opts) do
    case fetch_index() do
      {:ok, docs} ->
        by_url = Map.new(docs, &{&1.url, &1})

        {:ok, changed} =
          ReactiveDag.Node.Rows.reconcile(opts[:cell], Map.keys(by_url),
            upsert: fn url -> Map.get(by_url, url) end
          )

        {:ok, %{changed: changed, unreachable: []}}

      # an upstream we could not reach writes NOTHING — see below
      {:error, reason} ->
        {:ok, %{changed: [], unreachable: [{"docs", reason}]}}
    end
  end
end

upsert: returns the row you observed and the library writes it, deciding changed? against the declared fingerprint. Return nil for a key you could not observe and it is skipped — written nowhere, reported as nothing.

reconcile/3 also does the set math: keys the scan no longer found are retired (their row destroyed) and reported as changed, because something disappearing is a change your downstream nodes need to see.

Then poll and drain:

{:ok, results} = ReactiveDag.Source.poll_all(plan)
for {_source, %{changed: keys}} <- results do
  ReactiveDag.Graph.dirty_parents(plan, "docs", keys, ReactiveDag.Node.KeyRule)
end

ReactiveDag.Drain.run(plan, recompute: ..., key_rule: ...)

poll_all/1 finds every scanner from the plan's poll declarations, so there is no hand-kept list to fall out of date. graph/2 has already checked that each declared module implements the behaviour.

A source is a NODE, so everything reading it is an ordinary edge — one crawl whose rows belong to several downstream nodes needs no scan-specific machinery, just a consumer per part.

Two things a scanner must get right

fingerprint decides what "changed" means. A scanned row carries fields that move on every observation without the observation having changed: a last_seen_at by definition, an etag a server may re-issue for identical bytes. The default change detection compares every attribute, so without a fingerprint each poll fires the whole cascade. With fingerprint [:content_md5]:

first poll                      → changed: ["/a", "/b"]
re-poll, identical bytes        → changed: []          ← last_seen_at moved; nothing else did
re-poll, "/b" gone from index   → changed: ["/b"]      ← retired

Use the (row -> value) form when "the same observation" is not a plain field comparison — folding a listing title into the digest, say, so a renamed document re-fires even though its bytes are identical.

An upstream you could not reach writes nothing. If the fetch fails and the scan hands reconcile/3 an empty set, every key reads as vanished, every row is retired, and a downstream rollup over an empty set typically reads as vacuously fine. A scan that could not look must never render as a scan that found nothing — so on failure, write nothing, retire nothing, and report it in unreachable: so the host can surface the gap.

Human edits — a managed list or an approval writes a leaf like anything else: the host's normal write, then a dirty mark (or dirties_on).

More:Sources and scanning covers the parts a real scanner runs into — why the poll/drain split is a design invariant rather than a convention, one crawl feeding several downstream nodes (and {:skip, key}, which is how a projecting node declines what is not its own), the corollary when some upstreams fail and others succeed, choosing between dirties_on and a Source, and when a scanner is the wrong tool.

LLM and other expensive per-row work

An LLM recompute needs no library code. ash_ai's prompt/2 builds an ordinary generic Ash action, and per_key calls a generic action once per input row — so an LLM node is per_key with a prompt behind it. ash_ai is not a dependency of this library; hosts that want it add it themselves.

defmodule MyApp.Summaries do
  use Ash.Resource, data_layer: AshPostgres.DataLayer, extensions: [ReactiveDag.Node]

  import AshAi.Actions, only: [prompt: 2]

  attributes do
    attribute :key, :string, primary_key?: true
    attribute :summary, :string
    attribute :fingerprint, :string      # where the input hash lives
  end

  actions do
    defaults [:read, :destroy]
    create :upsert do upsert?(true); accept([:key, :summary, :fingerprint]) end

    action :summarise, :map do
      argument :text, :string, allow_nil?: false
      run prompt("openai:gpt-4o",
            prompt: {"You summarise documents", "Summarise: <%= @input.arguments.text %>"})
    end
  end

  reactive do
    recompute_by :key, to: :docs, from: :url

    per_key :summarise,
      args: [text: :body],          # the row's :body becomes the `text` argument
      fingerprint: [:body],         # SKIP the call when :body is unchanged
      into: [summary: :summary]     # the result's "summary" → this :summary

    context :people                 # settled context; never re-triggers
  end
end

The library drives the loop: scope to the claimed keys, read those rows, call the action once each, write the structured output through the payload loop.

fingerprint: is the line that matters, because the call costs money and latency. It partitions before calling, so an unchanged row costs nothing at all:

first pass                    → %{called: 1, skipped: 0}
second pass, nothing moved    → %{called: 0, skipped: 1}

Those counts come back in the drain report's step meta, so the saving is visible rather than assumed. A run action would be opaque — the library cannot know what it depends on — which is why fingerprinting lives on this rung.

Two more options worth knowing:

max_concurrency: — bounded Task.async_stream over the rows, since the latency is a remote call rather than CPU. Results apply in row order regardless, so the changed-key list stays deterministic.
context :people — the LLM reads a curated table but must not re-run every time someone edits it. A context edge is a real input (validated, ordered by depth so it settles first) that simply does not propagate.

Nothing here is LLM-specific: an embedding call, a PDF fetch, or an OCR pass is the same rung with a different action.

More:LLM nodes goes into the economics — why the drain being sequential means parallelism has to live inside a recompute, why embeddings are usually not a node at all, how dirty-key scoping bounds the bill, testing a prompt node without a model, and the rough edges worth knowing before you ship one.

Reading results

A node's results are its own rows, so the first answer is an ordinary Ash read:

MyApp.BudgetHealth |> Ash.Query.filter(status == "failing") |> Ash.read!()

For the same rows addressed by cell key — how the DAG talks about them — go through the cell:

cell = plan.cells["budget_health"]

ReactiveDag.Node.Rows.all(cell)               # [%{key:, status:, record:}]
ReactiveDag.Node.Rows.status_histogram(cell)  # %{"failing" => 1, "present" => 4}
ReactiveDag.Verdict.for_cell(cell)            # a rolled verdict + failing sample

ReactiveDag.Insights is the graph viewed from outside — levels/1 and edges/1 for structure, cell_status/2 and summary/1 for per-cell state, pending/1 for what the next drain would do. No UI dependency; reactive_dag_dashboard renders it.

Watching a drain

The drain emits :telemetry, so a dashboard, a metrics backend and a log can each observe it without any of them changing how it is called:

:telemetry.attach("drain-log", [:reactive_dag, :drain, :stop], fn _e, m, meta, _ ->
  Logger.info("drained #{length(meta.cells_touched)} cells in #{m.duration_us}us")
end, nil)

start / step / stop / exception. The step event carries the changed keys, not just a count — which is what lets a consumer refresh only what moved rather than re-reading the graph.

The seams

The engine decides when and in what order cells recompute. It never decides how, or what a value means. Three named seams take the domain:

ReactiveDag.RecomputeStrategy — how a cell recomputes. Node.Recompute handles everything above; a host with a different execution model (the compliance portal runs set-based SQL keyed on op) brings its own.
ReactiveDag.KeyRule — how a change propagates to a parent. Node.KeyRule reads it off the block; recompute_by usually means you never touch this.
ReactiveDag.Source — how outside state gets into a leaf, in a poll phase outside the drain so one unreachable vendor cannot wedge the rest.

A host can also skip the DSL entirely: build ReactiveDag.Cell structs, call ReactiveDag.Graph.build/1, and bring its own strategy. That is how both apps ran before adopting the Node surface, and the substrate still supports it — Cell's meta is an open map with an Access impl, so host fields ride along and read as cell[:field].

The library owns the schedule and one table. The host owns its resources, its op algebra, and its executor.

Design notes

A few decisions worth knowing, because each replaced something that seemed reasonable first.

Verdicts are rows. There used to be a tableless node shape (verdict? true) that wrote a status straight into a coordination table. It saved a migration when the answer was one word, and cost a ceiling: that table's schema was fixed, so the moment a verdict wanted a headroom the shape had nothing to offer and you wrote the table anyway. A row costs a migration and answers every later question.

There is no shadow table. The library used to write a coordination row per (cell, key), carrying a status and freshness. Once every node had a resource, that table recorded nothing the resource didn't already say — so it went, along with the writer seam that existed to extend it. A host that wants a source_ref or a last_seen_at puts it on the node's resource, where the rest of the row is.

Retirement destroys the row. A unit whose inputs have all gone produces nothing, and a derived row you cannot distinguish from a live one defeats the point of materializing it. So a vanished unit's row is destroyed and its key reported as changed, which is why a node that can retire needs a destroy action.

No command frontier. There was a second seq-ordered frontier for intents, with per-scope serialization and a human-in-the-loop blocked state. In both hosts the commands turned out to be straight CRUD drained inline, so nothing was ever queued; the database already provided the serialization, and its scope-freeze turned a failed edit into a wedged queue. A genuinely deferred, approval-gated write is the case that would justify bringing it back.

Status

Both hosts run on this substrate — a per-key Elixir recompute that calls LLMs and parses PDFs (cascade) and a set-based SQL recompute (the compliance portal) — which is the evidence that the seams are in the right places. See ADR-001 for the boundary and the design law behind it.