Honker.Queue (honker v0.1.4)

Copy Markdown

Named queue operations. All methods take a Honker.Database and the queue name as the first two args.

{:ok, id} = Honker.Queue.enqueue(db, "emails", %{to: "alice@example.com"})
{:ok, job} = Honker.Queue.claim_one(db, "emails", "worker-1")

Summary

Functions

Delete a pending or processing job by id. Returns {:ok, true} iff a row was removed. Idempotent on missing.

Atomically claim up to n jobs. Returns {:ok, [%Job{}, ...]}, possibly empty if the queue had no eligible rows.

Claim a single job or {:ok, nil} if the queue is empty.

Enqueue a job. payload is any term — it's JSON-encoded on the way in. Options

Enqueue inside an open transaction. The job is only visible to claimers after Honker.Transaction.commit/1.

Read a single job row by id. Returns {:ok, map} with the row fields, or {:ok, nil} on miss.

Sweep expired processing rows back to pending. Returns {:ok, rows_touched}.

Functions

cancel(database, job_id)

Delete a pending or processing job by id. Returns {:ok, true} iff a row was removed. Idempotent on missing.

IMPORTANT: cancel does NOT interrupt a worker currently running the handler. It invalidates the worker's claim — its next ack/heartbeat returns false. If you need the handler to actually halt, build that signal in your app.

claim_batch(db, queue_name, worker_id, n)

Atomically claim up to n jobs. Returns {:ok, [%Job{}, ...]}, possibly empty if the queue had no eligible rows.

claim_one(db, queue_name, worker_id)

Claim a single job or {:ok, nil} if the queue is empty.

enqueue(db, queue_name, payload, opts \\ [])

Enqueue a job. payload is any term — it's JSON-encoded on the way in. Options:

  • :delay — seconds from now before claimable
  • :run_at — absolute unix epoch (ignored if :delay is set)
  • :priority — higher = picked first within queue (default 0)
  • :expires — seconds from now; drops out of claim pool after

enqueue_tx(transaction, queue_name, payload, opts, queue_opts)

Enqueue inside an open transaction. The job is only visible to claimers after Honker.Transaction.commit/1.

queue_opts lets the caller pass per-queue options ([visibility_timeout_s: _, max_attempts: _]) when the transaction handle was obtained without going through a Honker.Database carrying a registered queue config. Pass [] to use the defaults (300s / 3 attempts).

get_job(database, job_id)

Read a single job row by id. Returns {:ok, map} with the row fields, or {:ok, nil} on miss.

sweep_expired(database, queue_name)

Sweep expired processing rows back to pending. Returns {:ok, rows_touched}.