ReactiveDag.Cascade (reactive_dag v0.17.0-rc.62)

Copy Markdown View Source

A CHANGE, propagated to completion — in one transaction, in memory, stopping only where it must.

Something writes a row. Everything downstream that can run, runs: the graph is walked shallowest-first, each cell recomputed and its parents queued, until either there is nothing left or the walk reaches a node that cannot be finished inline. That node's work is recorded as a suspension and the branch stops. The transaction closes AT the suspension, not around it — everything that ran is durable, and nothing slow is inside.

What this replaces, and why

The drain read a queue of CONCLUSIONS. A row (cell, key) meant "this cell needs recomputing", computed by walking the graph at mark time — so the conclusion aged between being written and being read, and the drain's job was to find work rather than to do it. Three costs followed, and one of them was a production failure: claim, recompute and propagate shared one transaction, so a nine-minute extraction held a connection until the database closed it.

A cascade starts from an explicit origin instead. It never asks the database what needs doing; it is told what changed and follows the consequences. The only thing the database holds is where it had to stop.

The walk

run(plan, origins)
  
   pop a pending cell with NO PENDING UPSTREAM
      merge everything else queued for it first
  
   suspends? yes> record one suspension per changed row
                       stop THIS branch; others carry on
  
   no > recompute in a savepoint
            queue each parent with the units this change touched

Three properties worth naming, because none held before:

  • A diamond recomputes its apex once. Two inputs of one cell changing in the same cascade merge before it runs. The queue could only manage this by luck — two marks coalesced if they happened to land before the claim.
  • Depth ordering is exact. In memory, not re-derived by a SELECT DISTINCT per pass.
  • A suspension truncates one branch. The rest of the graph keeps running and commits.

Failure

This changes what a failure costs, and the change is real. A drain claimed work with a DELETE before doing it, so a crash rolled back and the item returned: the queue remembered what was outstanding by still holding it.

A cascade holds nothing. A failure mid-walk rolls back the whole subtree and leaves no trace — the change is lost unless its source observes it again. That is acceptable because sources are idempotent and re-observe on a schedule: a crawl finding the same document with the same fingerprint writes nothing, and one finding it changed re-triggers the cascade. But the recovery story for fast work is now re-observation, not retry.

Slow work is different, and deliberately so: its suspension is committed before the job runs, so a crashed resumption resumes from a row that still exists.

Summary

Types

Where a cascade starts: a cell, the keys of it that changed, and the version recording each change.

Functions

Propagate origins through plan, in one transaction, to completion.

Types

origin()

@type origin() :: %{
  :cell => String.t(),
  :keys => [String.t()],
  optional(:versions) => %{required(String.t()) => String.t()},
  optional(:diffs) => map(),
  optional(:looped) => boolean()
}

Where a cascade starts: a cell, the keys of it that changed, and the version recording each change.

versions may be empty, and a key with no version yields a suspension carrying "*" — a resumption that recomputes the whole cell rather than the rows that moved.

Functions

run(plan, origins, opts \\ [])

@spec run(ReactiveDag.Plan.t(), [origin()], keyword()) ::
  {:ok, ReactiveDag.Report.t()}

Propagate origins through plan, in one transaction, to completion.

Returns the trace. report.suspended names every point where the cascade stopped, which is the only part of it the database also knows.

Options

  • :resumption_scheduler — how a recorded suspension becomes a job.

  • :max_feedback_passes — how many times ONE key may cross ONE declared feedback edge within this cascade before the walk raises RunawayError (else config :reactive_dag, max_feedback_passes:, else 3). A converging loop never reports the same key changed twice, so this binds only on an op whose change reporting is broken.

  • :max_feedback_laps — how many consecutive resumption-driven laps around a declared loop may suspend before recording the next suspension raises instead (else config :reactive_dag, max_feedback_laps:, else 20). This is the durable twin of :max_feedback_passes: a loop through a suspends cell ends every cascade cleanly, so the count rides on the suspension rows and ResumptionWorker hands it back via :feedback_lap.

  • :feedback_lap — the lap the resumed suspensions were recorded on. Supplied by ResumptionWorker; a host driving resumptions itself should pass the max lap of the suspensions it read, or oscillation through its suspending cells goes unbounded.