crystal-pg-queue

crystal-pg-queue is a hyper-lean, high-performance background job queue engine written in Crystal. Powered by PostgreSQL 18's native FOR UPDATE SKIP LOCKED transaction primitives, it eliminates the need for auxiliary state stores like Redis, Valkey, or RabbitMQ.

Features

Zero-Extra-Infrastructure: Operates entirely within your primary PostgreSQL 18 database.
Atomic Concurrency: Prevents race conditions across multiple worker processes using FOR UPDATE SKIP LOCKED.
Stateless Worker Pool: Supports configurable concurrency pools and queue prioritization.
JSON Job Serialization: Strongly-typed payload serialization across process and container boundaries.
GPLv3 Licensed: Released under the GNU General Public License v3.0.

Official Repository

The official repository for crystal-pg-queue is hosted on GitLab: https://gitlab.com/renich/crystal-pg-queue

Installation

Add crystal-pg-queue to your shard.yml:

dependencies:
  crystal-pg-queue:
    gitlab: renich/crystal-pg-queue
    version: ~> 0.1.0

Run shards install.

Usage

1. Define a Job

Inherit from PGQueue::Job and include JSON::Serializable:

require "crystal-pg-queue"

struct WelcomeEmailJob < PGQueue::Job
  include JSON::Serializable

  property user_id : Int64
  property email : String

  def initialize(@user_id, @email)
  end

  def perform(db : DB::Database)
    puts "[Job] Sending welcome email to #{email} (User ##{user_id})..."
  end
end

2. Register Job Deserializer & Worker Loop

require "db"
require "pg"
require "crystal-pg-queue"

# Connect to PostgreSQL 18
db = DB.open(ENV["DATABASE_URL"])

# Register job type for worker deserialization
PGQueue::Worker.register_job(WelcomeEmailJob)

# Enqueue job
job = WelcomeEmailJob.new(user_id: 42_i64, email: "user@example.com")
PGQueue::Client.enqueue(db, job, queue: "default")

# Start background worker loop
worker = PGQueue::Worker.new(db, queues: ["default"], pool_size: 4)
worker.start