Shigoto 仕事

PostgreSQL-backed background job processing for Erlang.

Shigoto (仕事, "work") is a feature-rich job queue built on PostgreSQL's FOR UPDATE SKIP LOCKED for safe multi-node operation. No Redis or external broker needed — if you have PostgreSQL, you have a job queue.

Features

Core

• PostgreSQL-backed — Jobs stored in PostgreSQL via pgo. Transactional enqueue using the same pool as your application.
• Multi-node safe — FOR UPDATE SKIP LOCKED ensures each job is claimed by exactly one node. No coordination required.
• Priority queues — Higher priority jobs are claimed first. Multiple queues with independent concurrency limits.
• Dynamic queues — Add and remove queues at runtime without restart.
• Exponential backoff — Failed jobs retry with min(attempt^4 + jitter, 1800) second delays, or provide a custom backoff/2 callback.

Scheduling

• Cron scheduling — Built-in 5-field cron parser with leader election via advisory locks. Catches up on missed intervals after restarts.
• Scheduled jobs — Enqueue jobs for future execution with scheduled_at.
• Job snoozing — Workers can return {snooze, Seconds} to reschedule without consuming a retry attempt.

Reliability

• Job dependencies — depends_on chains ensure jobs execute in order. Cycle detection prevents deadlocks.
• Batches — Group jobs with completion callbacks. Batch state tracks completed/discarded counts.
• Unique jobs — Prevent duplicates with configurable keys, states, time windows, debounce, and field replacement on conflict.
• Stale job rescue — Heartbeat-based detection of zombie jobs with automatic rescheduling.
• Graceful shutdown — Waits for in-flight jobs to finish before stopping.

Resilience (seki integration)

• Rate limiting — Per-worker token bucket, sliding window, GCRA, or leaky bucket.
• Circuit breaking — Auto-opens on repeated failures, configurable per worker.
• Bulkhead — Per-worker concurrency limits (local node) and global concurrency limits (across nodes via PostgreSQL).
• Load shedding — CoDel-based system-level protection, shedding low-priority jobs first.

Operations

• Encryption — AES-256-GCM encryption for job args at rest with key rotation support.
• Middleware — Composable before/after hooks for logging, metrics, authorization.
• Telemetry — 16+ telemetry events covering job lifecycle, queue operations, resilience, batches, and cron.
• Health check — shigoto:health/0 reports pool status, job counts, stale jobs, and queue health.
• Bulk operations — insert_all/1, cancel_by/2, retry_by/2 for batch operations.
• Auto-archival — Old jobs archived to shigoto_jobs_archive table, then pruned.
• Dashboard — shigoto_board provides a real-time web dashboard.

Testing

• DB-free worker runner — perform_job/2,3 runs a worker through the real perform path (middleware + deps_results) with no database.
• Enqueue assertions — assert_enqueued/1,2, refute_enqueued/1, all_enqueued/0,1 in shigoto_testing.
• Testing modes — inline (run on insert) / manual (capture without running), armed via a two-key confirmation so they can never silently activate in production.
• Synchronous drain — drain_queue/1 processes all jobs synchronously for integration tests.

Quick Start

Add to your deps:

{deps, [
    {shigoto, {git, "https://github.com/Taure/shigoto.git", {branch, "main"}}}
]}.

Configure in sys.config:

{shigoto, [
    {pool, my_app_db},
    {queues, [{<<"default">>, 10}, {<<"emails">>, 5}]},
    {poll_interval, 5000},
    {cron, [
        {<<"daily_cleanup">>, <<"0 3 * * *">>, my_cleanup_worker, #{}}
    ]}
]}

Run the migration:

shigoto_migration:up(my_app_db).

Define a worker:

-module(my_email_worker).
-behaviour(shigoto_worker).
-export([perform/1]).

perform(#{<<"to">> := To, <<"subject">> := Subject}) ->
    send_email(To, Subject),
    ok.

Enqueue jobs:

%% Simple insert
shigoto:insert(#{
    worker => my_email_worker,
    args => #{<<"to">> => <<"user@example.com">>, <<"subject">> => <<"Welcome">>}
}).

%% Scheduled for later
shigoto:insert(#{
    worker => my_cleanup_worker,
    scheduled_at => {{2026, 3, 20}, {3, 0, 0}}
}).

%% With priority and queue
shigoto:insert(#{
    worker => my_urgent_worker,
    args => #{},
    priority => 10,
    queue => <<"critical">>
}).

Job Lifecycle

available → executing → completed
                     ↘ retryable → available (retry with backoff)
                     ↘ discarded (max attempts reached)
                     ↘ snoozed → available (rescheduled, attempt preserved)

Jobs can also be cancelled via shigoto:cancel/2 and retried via shigoto:retry/2.

Worker Callbacks

All optional except perform/1:

Configuration

Supervision Tree

shigoto_sup (one_for_one)
  ├─ shigoto_executor_sup    — simple_one_for_one for job execution
  ├─ shigoto_queue_sup       — one gen_server per queue
  │    ├─ shigoto_queue:default
  │    └─ shigoto_queue:emails
  ├─ shigoto_cron            — cron scheduling with leader election
  ├─ shigoto_pruner          — hourly archival and cleanup
  ├─ shigoto_heartbeat       — periodic heartbeat updates
  ├─ shigoto_notifier        — LISTEN/NOTIFY (optional)
  └─ shigoto_stager          — surfaces newly-due jobs, leader election (optional)

Guides

• Getting Started
• Workers
• Cron Scheduling
• Batches
• Middleware
• Resilience
• Job Dependencies
• Encryption
• Testing

Ecosystem

• shigoto_board — Real-time Arizona LiveView dashboard
• opentelemetry_shigoto — OpenTelemetry instrumentation

Requirements

• Erlang/OTP 27+
• PostgreSQL 9.5+

License

MIT