@spec at(point()) :: [t()]

Every suspension at a stopping point, oldest first.

A resumption job's first act. An empty list is the ORDINARY outcome of a duplicate job — the work was already done and discharged — not an error.

Ordered by id, which is a UUIDv7 and therefore sorts by creation. That is not decorative: when several versions merge into one recompute, they must be applied in the order the changes happened.

@spec discharge([String.t()]) :: non_neg_integer()

Discharge suspensions BY ID, returning how many were removed.

By id, never by point — this is the line the append-only design rests on. A DELETE … WHERE tenant = … AND waiting = … would also remove suspensions written while the job was running, silently discarding changes nobody observed. Naming the ids the job actually read cannot do that.

Called in the same transaction as the resumption's writes, so a failed resumption leaves its suspensions in place and the work is retried.

@spec oban_table() :: String.t()

Oban's job table, validated. Configurable only so the SQL above can be exercised against a real database without colliding with a host's own table.

@spec pending?(keyword()) :: boolean()

Whether anything is suspended for this tenant.

@spec points(keyword()) :: [
  %{
    point: point(),
    reason: reason(),
    count: pos_integer(),
    oldest: DateTime.t()
  }
]

The distinct stopping points with work outstanding, for one tenant.

Non-consuming, and aggregated: one entry per point per reason, with how many suspensions have accumulated there and how long the oldest has waited. This is what an operator's view reads — "twelve things waiting on a person" — and what surfaces a point whose resumption keeps failing, since its count climbs while its oldest recedes.

@spec record(point(), String.t(), reason(), non_neg_integer()) :: String.t()

Record that a cascade stopped, and return the new row's id.

Called from INSIDE the cascade's transaction, so a rolled-back cascade leaves no suspension: the change it describes never happened.

version_id may be "*" when the change could not be attributed — a source that cannot say which of its items moved, or a write path that records no diff. Resumption then recomputes the whole cell: expensive, correct, and honest. This gives the whole-cell marker a principled meaning it has never had — not "something happened somewhere", but "this suspension could not be narrowed".

lap is how many consecutive resumption-driven trips around a declared feedback loop led to this suspension — 0 (the default, and the value for every suspension outside a loop) means the work arrived here from an external change. The column exists because a loop through a suspending cell ends every CASCADE cleanly: no in-memory counter survives the suspend → commit → resume chain, so the count must ride on the one thing that does — this row. ReactiveDag.Cascade refuses to record a lap past its budget, which is what bounds the chain.

@spec revive(keyword()) :: [integer()]

Make stranded resumption jobs fetchable again, returning the job ids revived.

Raises max_attempts above attempt — the same expression Oban's own retry_all_jobs/2 uses — and clears the backoff so the queue picks the job up on its next poll. Idempotent: a job already fetchable does not match.

Safe to run blind. A resumption whose suspensions were discharged in the meantime finds nothing at step 1 and exits, which is its ordinary duplicate path.

@spec savepoint((-> result)) :: result | {:error, term()} when result: term()

Run fun in a SAVEPOINT: if it returns {:error, reason}, everything it did is undone and {:error, reason} comes back — without disturbing the transaction around it.

This is what lets one fallible cell fail inside a cascade that is otherwise committing. The branch that failed stops; every other branch carries on and commits.

It must RETURN its failure, not raise

An exception inside a nested transaction aborts the OUTER one: Postgres marks the connection failed and every later statement errors until the whole thing rolls back. A savepoint only isolates a failure that arrives as a value.

That is why ReactiveDag.Source's poll/1 returns {:error, reason} "contained, not raised" — the contract was already the right shape for this.

A repo without transaction/2 runs fun directly: no savepoint, so a failure is not isolated.

@spec stranded(keyword()) :: [map()]

Stopping points whose resumption job can never run.

A job that exhausts its attempts is normally discarded, which is visible. But Oban returns a failed job to available with a backoff BEFORE checking attempts, so a final attempt that fails leaves attempt == max_attempts in state available — and the fetch query requires attempt < max_attempts (Oban.Engines.Basic). The job is then unfetchable and undiscarded: it will never run and nothing reports it as failed.

That would merely be a stalled point, except the resumption worker's uniqueness is states: :incomplete, which INCLUDES available. So the stranded job also dedups every future enqueue for its point. The suspension stays outstanding, no job will ever discharge it, and the queue looks healthy.

Observed in production: a resumption whose third attempt hit DBConnection.ConnectionError while the machine was being replaced.

Returns a point plus its job_id for each. Repair with revive/1. Note that Oban.retry_job/1 does NOT fix these — it skips jobs already in available (retry_all_jobs/2), so it reports success and changes nothing.

@spec table() :: String.t()

The suspension table's name, validated.

@spec tenant(keyword()) :: String.t()

The tenant a call is about: opts[:tenant], else "*" (untenanted).

"*" rather than nil because a null never equals itself. Every read here matches on the tenant, and tenant = NULL matches nothing — an untenanted suspension would be written and then never found, which is the quiet failure of tenancy in its worst form: a resumption that finds no work reports SUCCESS. "*" is also already this library's spelling for "the whole thing", so the vocabulary is not new.

Normalised in ONE place so a caller passing nil, omitting the option, or passing an atom all mean the same row.

@spec transaction((-> result)) :: result when result: term()

Run fun in a transaction — the cascade's transaction.

Bounded, unlike the drain's it replaces. The drain wrapped its recompute in timeout: :infinity and justified it with "a recompute legitimately runs for minutes" — which is exactly the condition that made a nine-minute extraction hold a connection until the database killed it.

A cascade transaction contains only fast work by construction: anything slow declared itself so and became a suspension. An infinite timeout here would hide the one bug this design exists to remove — a cell declared cheap that is not. Configure with:

config :reactive_dag, cascade_timeout: 30_000

A repo without transaction/2 runs fun directly. The library only ever needed query!/2, and requiring a new capability would break such a host on upgrade for a guarantee it may not want — at the cost that its cascades are not atomic.