shigoto (shigoto v1.9.10)
View SourcePublic API for the Shigoto background job system.
Shigoto (仕事, "work") is a PostgreSQL-backed job queue for the Nova ecosystem.
Jobs are claimed via FOR UPDATE SKIP LOCKED for safe multi-node operation.
Quick Start
%% Define a worker
-module(my_email_worker).
-behaviour(shigoto_worker).
-export([perform/1]).
perform(#{<<\"to\">> := To, <<\"subject\">> := Subject}) ->
send_email(To, Subject),
ok.%% Enqueue a job
shigoto:insert(#{
worker => my_email_worker,
args => #{<<\"to\">> => <<\"user@example.com\">>, <<\"subject\">> => <<\"Hello\">>}
}).Bulk Insert
shigoto:insert_all([
#{worker => my_worker, args => #{<<\"id\">> => 1}},
#{worker => my_worker, args => #{<<\"id\">> => 2}},
#{worker => my_worker, args => #{<<\"id\">> => 3}}
]).Batches
{ok, Batch} = shigoto:new_batch(#{
callback_worker => my_batch_callback,
callback_args => #{<<\"report\">> => 1}
}),
BatchId = maps:get(id, Batch),
shigoto:insert(#{worker => step1, args => #{}, batch => BatchId}),
shigoto:insert(#{worker => step2, args => #{}, batch => BatchId}).Transactional Enqueue
Enqueue jobs atomically with your own database writes. Jobs inserted inside
transaction/1 commit together with the surrounding work, and are dropped if it
rolls back — so a job is never orphaned by a failed transaction, nor lost after a
successful one.
shigoto:transaction(fun() ->
{ok, User} = my_app:create_user(Params),
{ok, _Job} = shigoto:insert(#{
worker => welcome_email_worker,
args => #{~"user_id" => maps:get(id, User)}
}),
User
end).Summary
Functions
Add a queue at runtime without restart.
Cancel a job by ID. Stops the job if it is executing on this node.
Cancel a job by ID. Also stops executing jobs on this node.
Cancel jobs matching a pattern. Uses the active transaction's pool, or the configured pool. Filters: worker, queue, tags, args.
Cancel jobs matching a pattern. Filters: worker, queue, tags, args.
Drain a queue synchronously. Useful for testing.
Drain a queue with options (e.g., #{timeout => 5000}).
Get a batch by ID.
Get a job by ID.
Health check. Returns ok with stats or error with details.
Insert a job with default options.
Insert a job with options. Params: worker, args, queue, priority, scheduled_at, max_attempts, unique, tags, batch.
Bulk insert multiple jobs with default options.
Bulk insert multiple jobs with options.
Create a new batch for grouping jobs.
Pause a queue by name — stops claiming new jobs.
Run a worker through Shigoto's real perform path with no database, for unit
tests. Returns the raw worker result. See shigoto_testing.
Like perform_job/2 with extra job context (deps_results, attempt, ...).
Remove a queue at runtime. Waits for in-flight jobs to finish.
Report job progress (0-100). Call from within a worker's perform/1.
Resume a paused queue by name.
Retry a discarded or cancelled job. Uses the active transaction's pool, or the configured pool.
Retry a discarded or cancelled job.
Retry all jobs matching a filter. Uses the active transaction's pool, or the configured pool. Filters: worker, queue, state, tags.
Retry all jobs matching a filter. Filters: worker, queue, state, tags.
Run Fun inside a database transaction on the Shigoto pool.
Like transaction/1, on a specific pool via #{pool => Pool}. Nested calls inherit the outer pool.
Functions
-spec add_queue(binary(), pos_integer()) -> {ok, pid()} | {error, term()}.
Add a queue at runtime without restart.
Cancel a job by ID. Stops the job if it is executing on this node.
Not transaction-aware: the process stop and job_cancelled telemetry fire
immediately, so this always uses the configured pool rather than an enclosing
transaction/1,2 pool.
Cancel a job by ID. Also stops executing jobs on this node.
-spec cancel_by(map()) -> {ok, non_neg_integer()} | {error, term()}.
Cancel jobs matching a pattern. Uses the active transaction's pool, or the configured pool. Filters: worker, queue, tags, args.
-spec cancel_by(atom(), map()) -> {ok, non_neg_integer()} | {error, term()}.
Cancel jobs matching a pattern. Filters: worker, queue, tags, args.
-spec drain_queue(binary()) -> ok.
Drain a queue synchronously. Useful for testing.
Drain a queue with options (e.g., #{timeout => 5000}).
Get a batch by ID.
Get a job by ID.
Health check. Returns ok with stats or error with details.
Insert a job with default options.
Insert a job with options. Params: worker, args, queue, priority, scheduled_at, max_attempts, unique, tags, batch.
Bulk insert multiple jobs with default options.
Bulk insert multiple jobs with options.
Create a new batch for grouping jobs.
-spec pause_queue(binary()) -> ok | {error, not_found}.
Pause a queue by name — stops claiming new jobs.
-spec perform_job(module(), map()) -> ok | {ok, term()} | {error, term()} | {snooze, pos_integer()}.
Run a worker through Shigoto's real perform path with no database, for unit
tests. Returns the raw worker result. See shigoto_testing.
-spec perform_job(module(), map(), map()) -> ok | {ok, term()} | {error, term()} | {snooze, pos_integer()}.
Like perform_job/2 with extra job context (deps_results, attempt, ...).
Remove a queue at runtime. Waits for in-flight jobs to finish.
Report job progress (0-100). Call from within a worker's perform/1.
-spec resume_queue(binary()) -> ok | {error, not_found}.
Resume a paused queue by name.
Retry a discarded or cancelled job. Uses the active transaction's pool, or the configured pool.
Retry a discarded or cancelled job.
-spec retry_by(map()) -> {ok, non_neg_integer()} | {error, term()}.
Retry all jobs matching a filter. Uses the active transaction's pool, or the configured pool. Filters: worker, queue, state, tags.
-spec retry_by(atom(), map()) -> {ok, non_neg_integer()} | {error, term()}.
Retry all jobs matching a filter. Filters: worker, queue, state, tags.
-spec transaction(fun(() -> Result)) -> Result when Result :: term().
Run Fun inside a database transaction on the Shigoto pool.
Jobs enqueued with insert/1,2 or insert_all/1,2 from within Fun commit
atomically with any other work done on the same pool: if Fun raises, the
transaction rolls back and no jobs are enqueued. job_inserted telemetry fires
only after a successful commit.
To roll back, raise an exception. If a statement inside Fun fails and aborts the
transaction, the commit is rejected and transaction_rolled_back is raised — no
jobs are enqueued and no telemetry fires. Returns the value of Fun.
Like transaction/1, on a specific pool via #{pool => Pool}. Nested calls inherit the outer pool.