shigoto_repo (shigoto v1.9.10)

View Source

All 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

archive_jobs(Pool, Days)

-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.

cancel_by(Pool, Filters)

-spec cancel_by(atom(), map()) -> {ok, non_neg_integer()} | {error, term()}.

Cancel jobs matching a pattern. Supports worker, queue, tags filters.

cancel_job(Pool, JobId)

-spec cancel_job(atom(), integer()) -> ok | {error, term()}.

Cancel a job. Resolves dependencies so dependent jobs aren't stuck.

claim_jobs(Pool, Queue, Limit)

-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.

claim_jobs_fair(Pool, Queue, Limit)

-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.

complete_job(Pool, JobId)

-spec complete_job(atom(), integer()) -> ok | {error, term()}.

Mark a job as completed with no result. Equivalent to complete_job/3 with null.

complete_job(Pool, JobId, Result)

-spec complete_job(atom(), integer(), term()) -> ok | {error, term()}.

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.

discard_job(Pool, JobId)

-spec discard_job(atom(), integer()) -> ok | {error, term()}.

Mark a job as discarded. Resolves dependencies so dependent jobs aren't stuck.

fail_job(Pool, Job, Reason, BackoffSeconds)

-spec fail_job(atom(), map(), term(), pos_integer()) -> ok | {error, term()}.

Record a job failure with structured error and configurable backoff.

fetch_fanout_jobs(Pool, Queue, WindowSeconds)

-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(Pool, Filters0)

-spec find_jobs(atom(), map()) -> {ok, [map()]} | {error, term()}.

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_due_cron_entries(Pool)

-spec get_due_cron_entries(atom()) -> {ok, [map()]} | {error, term()}.

Get cron entries that are due for scheduling.

get_job(Pool, JobId)

-spec get_job(atom(), integer()) -> {ok, map()} | {error, not_found | term()}.

Get a job by ID.

insert_all/3

-spec insert_all(atom(), [map()], map()) -> {ok, [map()]} | {error, term()}.

Bulk insert multiple jobs in a single SQL statement. Returns inserted jobs.

insert_job(Pool, JobParams0, Opts)

-spec insert_job(atom(), map(), map()) -> {ok, map()} | {ok, {conflict, map()}} | {error, term()}.

Insert a new job. Respects worker defaults and unique constraints. Validates dependency cycles.

prune_archive(Pool, Days)

-spec prune_archive(atom(), pos_integer()) -> {ok, non_neg_integer()} | {error, term()}.

Delete archived jobs older than given days. Bounds the archive table.

prune_jobs(Pool, Days)

-spec prune_jobs(atom(), pos_integer()) -> {ok, non_neg_integer()} | {error, term()}.

Delete completed/discarded jobs older than given days.

rescue_stale_jobs(Pool, StaleSeconds)

-spec rescue_stale_jobs(atom(), pos_integer()) -> {ok, non_neg_integer()} | {error, term()}.

Rescue stale executing jobs. Uses heartbeat if available, otherwise attempted_at.

resolve_dependencies(Pool, FinishedJobId)

-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.

retry_by(Pool, Filters)

-spec retry_by(atom(), map()) -> {ok, non_neg_integer()} | {error, term()}.

Retry all jobs matching a filter. Supports worker, queue, state, tags.

retry_job(Pool, JobId)

-spec retry_job(atom(), integer()) -> ok | {error, term()}.

Retry a discarded or cancelled job by resetting to available.

snooze_job(Pool, JobId, Seconds)

-spec snooze_job(atom(), integer(), pos_integer()) -> ok | {error, term()}.

Snooze a job — reschedule for later without counting as a failure.

update_progress(Pool, JobId, Progress)

-spec update_progress(atom(), integer(), 0..100) -> ok | {error, term()}.

Update job progress (0-100).

upsert_cron_entry(Pool, Entry)

-spec upsert_cron_entry(atom(), map()) -> ok | {error, term()}.

Upsert a cron entry.