The behaviour a module implements to run a workflow's scheduled work.
A workflow declares two kinds of work: automatic steps, which should run as
soon as a record occupies them, and timeouts, which should run once a deadline
has passed. Both arrive as AshWorkflow.Scheduler.Work structs. A scheduler
decides when each one runs, and how durably. It does not decide what
running means — see "The split" below.
Choosing one
workflow do
scheduler AshWorkflow.Scheduler.Oban, queue: :workflow, check_interval: "* * * * *"
endOr once for an application:
config :ash_workflow, scheduler: {AshWorkflow.Scheduler.Oban, queue: :workflow}The split
The scheduler owns when and how durably. AshWorkflow owns what happens:
which action runs, where a failure routes, what actor and authorization apply.
That division is why execute/3 lives here rather than in each
implementation. Two schedulers must not disagree about whether on_error
fires, so neither of them gets to decide.
So an implementation's job is to arrange for execute/3 to be called at the
right moment, and to survive a node restart while it waits.
Two strategies, one behaviour
The interesting constraint is that the sensible implementations work in opposite directions.
A discovering scheduler periodically asks the data layer which records
match, and nothing is recorded per deadline. That is what the Oban
implementation does, and its floor is the polling interval — one minute for
cron. It reads Work.match and ignores deadline_changed/2.
A registering scheduler is told each deadline as it becomes known and arms
a timer for it. Its floor is the timer, which is microseconds. It reads
Work.deadline and implements deadline_changed/2.
Neither is forced into the other's shape: deadline_changed/2 is optional, so
a discovering scheduler simply does not implement it, and a registering one
gets the callback it needs without every workflow paying for a table.
Callbacks
transform/3 is the only required callback, and the only one that runs at
compile time. It receives the resource's DSL state, every Work the workflow
declared, and the options given alongside the module in the DSL, and returns
the DSL state with whatever the implementation needs added — Oban triggers, an attribute to hold a deadline, nothing at all.
use AshWorkflow.Scheduler supplies a no-op default.
The rest are optional and run at runtime:
child_spec/1— a supervised process, for an implementation that holds state such as an in-memory timeline.deadline_changed/2— called after a record's state changes, so a registering implementation can re-arm. Given the record and the work still ahead of it.cancel/2— called when a record leaves a step, so a pending timer can be dropped. A discovering implementation needs neither.
Summary
Callbacks
Drop anything pending for this record.
A supervised child, for an implementation that needs a process.
A record's state changed, so the work ahead of it may have.
The shortest deadline this implementation can honour, in milliseconds.
Add whatever the implementation needs to the resource, at compile time.
Functions
The scheduler configured for an application, when a resource does not name one.
The instant this work becomes eligible for a record, or nil for work that is
eligible as soon as the record occupies the step.
Run one unit of work against one record.
Tell the resource's scheduler that a record's state changed.
The shortest deadline the given scheduler can honour, in milliseconds.
Callbacks
@callback cancel(record :: Ash.Resource.record(), [AshWorkflow.Scheduler.Work.t()]) :: :ok
Drop anything pending for this record.
@callback child_spec(opts :: keyword()) :: Supervisor.child_spec()
A supervised child, for an implementation that needs a process.
Returned to the host application to place in its own supervision tree.
@callback deadline_changed(record :: Ash.Resource.record(), [ AshWorkflow.Scheduler.Work.t() ]) :: :ok
A record's state changed, so the work ahead of it may have.
works is what remains ahead of the record in its new state. A registering
implementation re-arms from this; a discovering one does not implement it.
@callback precision_floor_ms(opts :: keyword()) :: pos_integer()
The shortest deadline this implementation can honour, in milliseconds.
AshWorkflow.Verifiers.ValidateTimeoutPrecision rejects a timeout shorter
than this, so that the DSL cannot promise a deadline the scheduler will miss
by more than the deadline itself. An implementation that does not define it
is assumed to poll once a minute, which is cron's floor.
@callback transform( dsl :: Spark.Dsl.t(), [AshWorkflow.Scheduler.Work.t()], opts :: keyword() ) :: {:ok, Spark.Dsl.t()} | {:error, term()}
Add whatever the implementation needs to the resource, at compile time.
Called by AshWorkflow.Transformers.AddScheduler with every Work the
workflow declares. Return the DSL state unchanged if nothing is needed.
Functions
The scheduler configured for an application, when a resource does not name one.
@spec due_at(AshWorkflow.Scheduler.Work.t(), Ash.Resource.record()) :: DateTime.t() | nil
The instant this work becomes eligible for a record, or nil for work that is
eligible as soon as the record occupies the step.
A registering implementation arms its timer from this.
@spec execute(AshWorkflow.Scheduler.Work.t(), Ash.Resource.record(), keyword()) :: {:ok, Ash.Resource.record()} | {:error, term()} | {:retry, non_neg_integer()}
Run one unit of work against one record.
This is what an implementation calls when it decides the moment has come. It
runs the work's action, and routes a failure to the step's on_error action
when one is declared.
Owned by AshWorkflow rather than by each implementation so that swapping the scheduler cannot change what a timeout does — only when it happens.
:attempt is the attempt number about to run, defaulting to 1. When the
action fails and attempt < work.retry.max_attempts, this returns
{:retry, delay_ms} instead of running on_error, where delay_ms comes
from AshWorkflow.Entities.Retry.delay_ms/2. The caller is responsible for
running the work again after that delay — execute/3 only decides that a
retry should happen, not when. Once the final attempt has failed, this
behaves exactly as it always has: on_error runs if declared, otherwise the
error is returned. With the default max_attempts of 1, {:retry, _} is
never returned.
@spec notify_state_change(Ash.Resource.record()) :: :ok
Tell the resource's scheduler that a record's state changed.
Called from AshWorkflow.Changes.RecordEvent after every state change. Works
out which work now lies ahead of the record and hands it to
deadline_changed/2, or to cancel/2 when the record has reached a step
with nothing left to schedule.
A scheduler that discovers work by polling implements neither callback, so this is a no-op for it.
@spec precision_floor_ms({module(), keyword()}) :: pos_integer()
The shortest deadline the given scheduler can honour, in milliseconds.
Falls back to one minute, cron's floor, for an implementation that does not
define precision_floor_ms/1.