Simple parent/child batch workflows for fan-out/fan-in.
The parent job moves to :waiting_for_children without blocking a BEAM process.
When the batch completes, an explicit continuation job is enqueued.
Examples
{:ok, parent} = Kathikon.Storage.insert(parent_job)
{:ok, batch} =
Kathikon.Batch.start(parent.id, [
{ChildWorker, %{"id" => 1}, [queue: :default]},
{ChildWorker, %{"id" => 2}, [queue: :default]}
], on_complete: {ReportWorker, %{"parent_id" => parent.id}})
{:ok, %{status: :running}} = Kathikon.Batch.status(batch.batch_id)See docs/batches.md.
Summary
Functions
Lists child job ids for a batch.
Returns results for completed child jobs in a batch.
Retries failed children in a batch.
Starts a batch from a parent job, enqueueing child jobs.
Returns batch status by batch id.
Types
Functions
Lists child job ids for a batch.
Examples
{:ok, child_ids} = Kathikon.Batch.children(batch_id)Returns results for completed child jobs in a batch.
Examples
{:ok, results} = Kathikon.Batch.results(batch_id)
Enum.filter(results, &(&1.state == :completed))@spec retry_failed(String.t()) :: {:ok, [Kathikon.Job.t()]} | {:error, term()}
Retries failed children in a batch.
Examples
{:ok, retried} = Kathikon.Batch.retry_failed(batch_id)
length(retried)@spec start(String.t(), [child_spec()], keyword()) :: {:ok, map()} | {:error, term()}
Starts a batch from a parent job, enqueueing child jobs.
Options
:on_complete—{WorkerModule, args}continuation when batch succeeds:success_policy—:all_succeeded(default),{:at_least, n}, or:allow_partial:queue— queue for child jobs
Examples
{:ok, batch} =
Kathikon.Batch.start(parent_job_id, [
{ProcessRowWorker, %{"row" => 1}, []},
{ProcessRowWorker, %{"row" => 2}, []}
], on_complete: {SummarizeWorker, %{}})Returns batch status by batch id.
Examples
{:ok, batch} = Kathikon.Batch.status(batch_id)
batch.status
#=> :running