BEAM-native durable job queue and task execution platform.
Examples
# Enqueue work
{:ok, job} = Kathikon.insert(MyApp.EmailWorker, %{"to" => "user@example.com"})
# Schedule for later
{:ok, job_id} = Kathikon.schedule(MyApp.DigestWorker, %{}, in: 3600)
# Inspect
{:ok, %{state: :completed}} = Kathikon.status(job.id)See the README and docs/ guides for v0.2.0 features:
atomic claiming, job history, dead-letter queue, scheduling, batches,
management APIs, and reporting.
Summary
Functions
Lists all jobs currently in storage.
Cancels a pending job.
Returns child jobs for a parent or batch id.
Atomically claims a specific job by id and transitions it to :running.
Atomically claims up to :limit available jobs from a queue (default 1).
Lists dead-letter jobs.
Discards a dead-letter job permanently.
Returns errors recorded for a job.
Fetches a job by id.
Returns durable history events for a job.
Inserts a job into the durable queue.
Creates a linked rerun of a job (original is unchanged).
Returns the persisted result for a completed job.
Retries a retryable, failed, or dead job in place.
Retries a dead-letter job by creating a linked rerun.
Schedules a job at a datetime, after a duration, or with a cron expression.
Ensures a dispatcher is running for the given queue.
Returns job status suitable for dashboards.
Functions
@spec all() :: [Kathikon.Job.t()]
Lists all jobs currently in storage.
Intended for tests, debugging, and reporting — not for hot paths in production.
Examples
Kathikon.all()
|> Enum.count(&(&1.state == :completed))
@spec cancel(String.t(), term()) :: {:ok, Kathikon.Job.t()} | {:error, term()}
Cancels a pending job.
Only jobs in cancellable states (:scheduled, :available, :retryable,
:claimed) succeed. Running jobs return {:error, :executing}.
Examples
{:ok, job} = Kathikon.cancel(job_id)
job.state
#=> :cancelled
{:ok, job} = Kathikon.cancel(job_id, :user_requested)
{:error, :executing} = Kathikon.cancel(running_job_id)
@spec children(String.t()) :: {:ok, [Kathikon.Job.t()]} | {:error, term()}
Returns child jobs for a parent or batch id.
Examples
{:ok, children} = Kathikon.children(parent_job_id)
Enum.map(children, & &1.state)
@spec claim( String.t(), keyword() ) :: {:ok, Kathikon.Job.t()} | {:error, term()}
Atomically claims a specific job by id and transitions it to :running.
Intended for external runners and management tools. Normal application code should let dispatchers claim jobs automatically.
Options
:node— claimant node (defaultnode()):pid— claimant pid (defaultself()):dispatcher_id— optional dispatcher reference
Examples
{:ok, job} = Kathikon.claim(job_id)
# perform work, then complete via storage callbacks or worker return
{:error, :not_claimable} = Kathikon.claim(already_running_id)
@spec claim_available( atom(), keyword() ) :: {:ok, [Kathikon.Job.t()]} | {:error, term()}
Atomically claims up to :limit available jobs from a queue (default 1).
Each returned job is already in :running state.
Examples
{:ok, [job]} = Kathikon.claim_available(:imports)
{:ok, jobs} = Kathikon.claim_available(:imports, limit: 5)
@spec dead_jobs(keyword()) :: {:ok, [Kathikon.Job.t()]} | {:error, term()}
Lists dead-letter jobs.
Options
:queue— filter by queue
Examples
{:ok, dead} = Kathikon.dead_jobs()
{:ok, dead} = Kathikon.dead_jobs(queue: :default)
@spec discard_dead(String.t(), term()) :: {:ok, Kathikon.Job.t()} | {:error, term()}
Discards a dead-letter job permanently.
Examples
{:ok, job} = Kathikon.discard_dead(dead_job_id)
job.state
#=> :discarded
{:ok, job} = Kathikon.discard_dead(dead_job_id, :no_longer_needed)
Returns errors recorded for a job.
Examples
{:ok, errors} = Kathikon.errors(job_id)
List.last(errors)
#=> %{"at" => ~U[...], "error" => "...", "attempt" => 1}
@spec fetch(String.t()) :: {:ok, Kathikon.Job.t()} | {:error, term()}
Fetches a job by id.
Examples
{:ok, job} = Kathikon.fetch(job_id)
job.state
{:error, :not_found} = Kathikon.fetch("missing")
Returns durable history events for a job.
Examples
{:ok, events} = Kathikon.history(job_id)
Enum.map(events, & &1.event)
#=> [:inserted, :claimed, :started, :completed]
@spec insert(module(), map(), keyword()) :: {:ok, Kathikon.Job.t()} | {:error, term()}
Inserts a job into the durable queue.
Starts the target queue dispatcher if needed. Emits [:kathikon, :job, :inserted].
Options
:queue— target queue (default:default):priority— higher values claim first (default0):max_attempts— retry limit (default from config):schedule_in— seconds until the job becomes available:schedule_at—DateTimeorNaiveDateTimewhen the job becomes available
Examples
{:ok, job} = Kathikon.insert(MyApp.EmailWorker, %{"to" => "user@example.com"})
{:ok, job} =
Kathikon.insert(MyApp.EmailWorker, %{"to" => "user@example.com"},
queue: :emails,
priority: 5,
max_attempts: 10
)
{:ok, job} = Kathikon.insert(MyApp.DigestWorker, %{}, schedule_in: 3600)
@spec rerun( String.t(), keyword() ) :: {:ok, Kathikon.Job.t()} | {:error, term()}
Creates a linked rerun of a job (original is unchanged).
The new job references the original via rerun_of and original_job_id.
Examples
{:ok, original} = Kathikon.fetch(dead_job_id)
{:ok, rerun} = Kathikon.rerun(original.id)
rerun.rerun_of
#=> original.id
Returns the persisted result for a completed job.
Requires config :kathikon, result: :store (the default).
Examples
{:ok, result} = Kathikon.result(completed_job_id)
{:error, {:invalid_state, :running}} = Kathikon.result(running_job_id)
@spec retry( String.t(), keyword() ) :: {:ok, Kathikon.Job.t()} | {:error, term()}
Retries a retryable, failed, or dead job in place.
Resets availability and increments nothing on the original record — the same job id is retried.
Examples
{:ok, job} = Kathikon.retry(retryable_job_id)
job.state
#=> :available
@spec retry_dead( String.t(), keyword() ) :: {:ok, Kathikon.Job.t()} | {:error, term()}
Retries a dead-letter job by creating a linked rerun.
Alias for rerun/2.
Examples
{:ok, job} = Kathikon.retry_dead(dead_job_id)
Schedules a job at a datetime, after a duration, or with a cron expression.
Returns {:ok, job_id} for one-time schedules or {:ok, schedule_id} for cron.
Options
:at—DateTimeorNaiveDateTimefor a one-time run:in— seconds from now for a one-time run:cron— 5-field cron expression for recurring schedules:queue,:priority,:max_attempts— same asinsert/3
Requires config :elixir, :time_zone_database, Tzdata.TimeZoneDatabase for
naive datetimes and cron.
Examples
{:ok, job_id} = Kathikon.schedule(MyApp.DigestWorker, %{}, in: 3600)
at = DateTime.add(DateTime.utc_now(), 5, :second)
{:ok, job_id} = Kathikon.schedule(MyApp.ReportWorker, %{}, at: at)
{:ok, schedule_id} =
Kathikon.schedule(MyApp.DigestWorker, %{"list" => "all"}, cron: "0 9 * * *")See Kathikon.Scheduler and docs/scheduling.md.
@spec start_queue(atom()) :: :ok
Ensures a dispatcher is running for the given queue.
Called automatically by insert/3 and schedule/3. Use when starting
queues dynamically.
Examples
:ok = Kathikon.start_queue(:imports)
Returns job status suitable for dashboards.
Examples
{:ok, status} = Kathikon.status(job_id)
status.state
#=> :completed
status.history_count
#=> 4