Behaviour for Shigoto job workers. Implement perform/1 to handle jobs.

The perform/1 callback receives the job's args map and should return ok on success or {error, Reason} on failure. Failed jobs are retried with configurable backoff up to max_attempts.

Workflow Context

Return {ok, Result} to store a JSON-encodable Result on the job. Any job that lists this one in its depends_on can then read that value. Plain ok stores a JSON null. Keep Result small (ids, status, a summary) — it is copied into every dependent's deps_results, so large payloads fan out.

When a job has dependencies, its args map carries a deps_results key holding the predecessors' results, keyed by predecessor job ID:

perform(#{~"x" := X, deps_results := #{ParentId := ParentResult}}) ->
    ...

The map shape is #{PredId :: integer() => Result :: term()}. A predecessor that returned plain ok (or was discarded/cancelled) contributes null. The deps_results key is absent for jobs with no dependencies, so existing workers are unaffected.

Optional Callbacks

Workers can declare defaults via optional callbacks:

• max_attempts/0 — Maximum retry attempts (default: 3)
• queue/0 — Default queue name (default: <<\"default\">>)
• priority/0 — Default priority (default: 0)
• timeout/0 — Execution timeout in ms (default: 300000 / 5 min)
• unique/0 — Uniqueness constraints (see below)
• tags/0 — Default tags for this worker
• backoff/2 — Custom backoff strategy (attempt, error -> seconds)
• rate_limit/0 — Per-worker rate limit config for seki
• concurrency/0 — Max concurrent executions (local, per-node via seki bulkhead)
• global_concurrency/0 — Max concurrent executions across all nodes (via PostgreSQL)
• middleware/0 — Worker-specific middleware list
• on_discard/2 — Called when a job is permanently discarded
• circuit_breaker/0 — Per-worker circuit breaker config

These are used as defaults when inserting jobs. Per-insert options in the params map always take precedence.

Uniqueness

Implement unique/0 to prevent duplicate jobs:

unique() ->
    #{
        keys => [worker, args],     %% Fields to check (worker, args, queue)
        states => [available, executing, retryable],  %% States to check against
        period => 300               %% Seconds (or infinity)
    }.

Rate Limiting

Implement rate_limit/0 to limit job execution rate:

rate_limit() ->
    #{
        limit => 100,           %% Max requests per window
        window => 60000,        %% Window in milliseconds
        algorithm => sliding_window  %% token_bucket | sliding_window | gcra | leaky_bucket
    }.

Custom Backoff

Implement backoff/2 to control retry delays:

backoff(Attempt, _Error) ->
    min(Attempt * 10, 300).  %% Linear backoff, max 5 minutes

Example

-module(cleanup_worker).
-behaviour(shigoto_worker).
-export([perform/1, max_attempts/0, queue/0, timeout/0, unique/0]).

perform(#{<<\"days\">> := Days}) ->
    delete_old_records(Days),
    ok.

max_attempts() -> 5.
queue() -> <<\"maintenance\">>.
timeout() -> 60000. %% 1 minute
unique() -> #{keys => [worker, args], period => 300}.