shigoto_repo (shigoto v1.9.10)
View SourceAll SQL operations for Shigoto jobs. Uses FOR UPDATE SKIP LOCKED
for safe multi-node job claiming.
Summary
Functions
Archive completed/discarded/cancelled jobs older than given days. Moves to archive table then deletes.
Cancel jobs matching a pattern. Supports worker, queue, tags filters.
Cancel a job. Resolves dependencies so dependent jobs aren't stuck.
Claim up to N available jobs from a queue using FOR UPDATE SKIP LOCKED.
Claim jobs with fair partition interleaving. One job per partition first, then fill.
Mark a job as completed with no result. Equivalent to complete_job/3 with null.
Mark a job as completed and store its JSON-encodable result on the row.
Mark a job as discarded. Resolves dependencies so dependent jobs aren't stuck.
Record a job failure with structured error and configurable backoff.
Fetch recent jobs for fanout queues. No locking — all nodes read the same rows.
Find jobs matching a filter map. Read-only; used by the testing helpers' DB-backed enqueue assertions.
Get cron entries that are due for scheduling.
Get a job by ID.
Bulk insert multiple jobs in a single SQL statement. Returns inserted jobs.
Insert a new job. Respects worker defaults and unique constraints. Validates dependency cycles.
Delete archived jobs older than given days. Bounds the archive table.
Delete completed/discarded jobs older than given days.
Rescue stale executing jobs. Uses heartbeat if available, otherwise attempted_at.
Resolve dependencies of a finished job.
Retry all jobs matching a filter. Supports worker, queue, state, tags.
Retry a discarded or cancelled job by resetting to available.
Snooze a job — reschedule for later without counting as a failure.
Update job progress (0-100).
Upsert a cron entry.
Functions
-spec archive_jobs(atom(), pos_integer()) -> {ok, non_neg_integer()} | {error, term()}.
Archive completed/discarded/cancelled jobs older than given days. Moves to archive table then deletes.
-spec cancel_by(atom(), map()) -> {ok, non_neg_integer()} | {error, term()}.
Cancel jobs matching a pattern. Supports worker, queue, tags filters.
Cancel a job. Resolves dependencies so dependent jobs aren't stuck.
-spec claim_jobs(atom(), binary(), pos_integer()) -> {ok, [map()]} | {error, term()}.
Claim up to N available jobs from a queue using FOR UPDATE SKIP LOCKED.
-spec claim_jobs_fair(atom(), binary(), pos_integer()) -> {ok, [map()]} | {error, term()}.
Claim jobs with fair partition interleaving. One job per partition first, then fill.
Mark a job as completed with no result. Equivalent to complete_job/3 with null.
Mark a job as completed and store its JSON-encodable result on the row.
The result is propagated to any jobs that depend on this one (see
resolve_dependencies/2) so they can read it via deps_results.
Mark a job as discarded. Resolves dependencies so dependent jobs aren't stuck.
-spec fail_job(atom(), map(), term(), pos_integer()) -> ok | {error, term()}.
Record a job failure with structured error and configurable backoff.
-spec fetch_fanout_jobs(atom(), binary(), pos_integer()) -> {ok, [map()]} | {error, term()}.
Fetch recent jobs for fanout queues. No locking — all nodes read the same rows.
Find jobs matching a filter map. Read-only; used by the testing helpers' DB-backed enqueue assertions.
Filters: worker, queue, args (jsonb containment), tags (array
containment), and state (a state binary or a list of them). An empty filter
returns all jobs. Args are decrypted so callers can match on plaintext.
Get cron entries that are due for scheduling.
Get a job by ID.
Bulk insert multiple jobs in a single SQL statement. Returns inserted jobs.
Insert a new job. Respects worker defaults and unique constraints. Validates dependency cycles.
-spec prune_archive(atom(), pos_integer()) -> {ok, non_neg_integer()} | {error, term()}.
Delete archived jobs older than given days. Bounds the archive table.
-spec prune_jobs(atom(), pos_integer()) -> {ok, non_neg_integer()} | {error, term()}.
Delete completed/discarded jobs older than given days.
-spec rescue_stale_jobs(atom(), pos_integer()) -> {ok, non_neg_integer()} | {error, term()}.
Rescue stale executing jobs. Uses heartbeat if available, otherwise attempted_at.
-spec resolve_dependencies(atom(), integer()) -> {ok, non_neg_integer()} | {error, term()}.
Resolve dependencies of a finished job.
Removes the finished job ID from every dependent's depends_on array (which
ungates the dependent once its last predecessor clears) and records the
finished job's result under its ID in each dependent's deps_results. A job
that produced no result (plain completion, discard, or cancel) contributes a
JSON null.
-spec retry_by(atom(), map()) -> {ok, non_neg_integer()} | {error, term()}.
Retry all jobs matching a filter. Supports worker, queue, state, tags.
Retry a discarded or cancelled job by resetting to available.
-spec snooze_job(atom(), integer(), pos_integer()) -> ok | {error, term()}.
Snooze a job — reschedule for later without counting as a failure.
Update job progress (0-100).
Upsert a cron entry.