Documentation

Run a hasql session against the pool, retrying once on cached plan errors.

Lock and fetch the next available job. In case no job is available returns Nothing.

The lock is set on the job row in an atomic way.

The job status is set to JobStatusRunning, lockedBy will be set to the worker id and the attemptsCount is incremented.

Example: Locking a SendMailJob

let workerId :: UUID = "faa5ba30-1d76-4adf-bf01-2d1f95cddc04"
job <- fetchNextJob @SendMailJob pool workerId

After you're done with the job, call jobDidFail or jobDidSucceed to make it available to the queue again.

Shared WHERE condition for fetching pending jobs as a SQL text fragment. Matches jobs that are either not started or in retry state, not locked, and whose run_at time has passed. Enum values are inlined as SQL string literals (PostgreSQL casts them to job_status).

Calls a callback every time something is inserted, updated or deleted in a given database table.

In the background this function creates database triggers and checks that they exist. The triggers notify this function about table changes using pg_notify. If installing the triggers fails temporarily, jobs are still found by the poller and trigger installation is retried.

This function returns a Async. Call cancel on the async to stop watching the database.

Example:

watchInsertOrUpdateTable "projects" do
    putStrLn "Something changed in the projects table"

Now insert something into the projects table. E.g. by running make psql and then running INSERT INTO projects (id, name) VALUES (DEFAULT, 'New project'); You will see that "Something changed in the projects table" is printed onto the screen.

Like watchForJob but allows disabling the poller-side trigger integrity check.

Periodically checks the queue table for open jobs. Calls the callback if there are any.

watchForJob only catches jobs when something is changed on the table. When a job is scheduled with a runAt in the future, and no other operation is happening on the queue, the database triggers will not run, and so watchForJob cannot pick up the job even when runAt is now in the past.

This function returns a Async. Call cancel on the async to stop polling the database.

Returns a SQL script to create the notification trigger.

The function body is always updated via CREATE OR REPLACE FUNCTION, which only locks the function's row in pg_proc, not the job table.

The trigger DDL is only executed when a trigger is missing: DROP TRIGGER takes an AccessExclusiveLock on the job table, which conflicts with every other lock — even the AccessShareLocks held by a running pg_dump. Running DROP + CREATE TRIGGER unconditionally on every start meant a job worker (re)started while a backup was in flight would block on the trigger DDL, and all subsequent INSERTs/UPDATEs on the job table would queue up behind that pending lock request, stalling job processing until the backup finished.

When a trigger is missing (first install or after make db), the table lock required by CREATE TRIGGER is acquired with NOWAIT. This prevents the repair from queuing behind an active writer and then blocking later writes. The caller falls back to the poller and retries later. Healthy triggers are never dropped.

The non-blocking advisory lock ensures only one process attempts installation at a time. Other processes immediately fall back to polling instead of queuing behind it. Changes to an existing trigger definition require an explicit migration because worker startup deliberately never drops a healthy, existing trigger.

Retuns the event name of the event that the pg notify trigger dispatches

Called when a job failed. Sets the job status to JobStatusFailed or JobStatusRetry (if more attempts are possible) and resets lockedBy

Called when a job succeeded. Sets the job status to JobStatusSucceded and resets lockedBy

Compute the delay before the next retry attempt.

For LinearBackoff, the delay is constant. For ExponentialBackoff, the delay doubles each attempt, capped at 24 hours.

Recover stale jobs that have been in JobStatusRunning for too long, likely due to a worker crash.

Two-tier recovery: - Recently stale jobs (within 24h) are set back to retry - Ancient stale jobs (older than 24h) are marked as failed

Parses a Text value to a JobStatus. Used by hasql decoders. Uses HashMap for O(1) lookup.

Non-blocking write to a TBQueue. Returns True if the value was written, False if the queue was full.