wagoe.jobs.shell.adapters.db — com.wagoe/wagoe-jobs 1.1.0-rc-1

Database-backed job queue (IJobQueue) on next.jdbc.

Lets an app run durable background jobs on its existing SQL database (H2, PostgreSQL or MySQL) without standing up Redis, and survive a Redis outage. Implements the same reliable-dequeue contract as the Redis adapter:

• dequeue-job! atomically claims the next ready row (sets status=processing, locked_by/locked_at) so a worker that crashes mid-job does not lose it.
• ack-job! removes the row once the worker is done with it.
• reclaim-abandoned-jobs! returns rows whose lease has expired (locked_at older than lease-ms) back to ready, i.e. at-least-once.

Portable SQL only (no SELECT ... FOR UPDATE SKIP LOCKED): the claim is a SELECT of the best candidate followed by a conditional UPDATE ... WHERE status='ready'. Concurrent workers that pick the same row have their UPDATEs serialized by the row lock — exactly one wins (update count 1); the losers retry the next candidate. This runs identically on H2, PostgreSQL and MySQL, so the reliability tests run on the default in-memory H2 test database.

The authoritative job value is the transit payload column (same wire format as the Redis adapter, wagoe.jobs.shell.adapters.codec). The indexed columns (queue, priority_rank, status, execute_at, locked_at) exist only for ordering, claiming, and reclaim.

The worker renews the lease while a handler runs (ports/ILeasedJobQueue), so lease-ms bounds how long a crashed worker's job waits, not how long a job may run (BOU-823).

Caveats:

• payload is VARCHAR(1000000) (LONGTEXT on MySQL); keep serialized jobs under ~1 MB.

create-db-idempotency-store clj

(create-db-idempotency-store ds)

An idempotency store on ds. Call create-recurring-tables! first.

create-db-job-queue clj

(create-db-job-queue ds & {:keys [lease-ms] :or {lease-ms default-lease-ms}})

Create a DB-backed job queue over the given next.jdbc datasource ds. :lease-ms (default 60s) is the in-flight lease after which a claimed job is reclaimable. Call create-jobs-table! once before use (or run the migration). Ids go as text, so on PostgreSQL ds must be the platform's, whose URL sets stringtype=unspecified.

create-db-job-store clj

(create-db-job-store ds)

Create a DB-backed job store over the given next.jdbc datasource ds. Call create-job-store-table! once before use (or run the migration).

create-db-recurring-state clj

(create-db-recurring-state ds)

Recurring-job state on ds, shared by every replica on it. Call create-recurring-tables! first.

create-job-store-table! clj

(create-job-store-table! ds)

Create the job_store table + its indexes if absent, on H2, PostgreSQL or MySQL. Call once at startup, or ship as a migration.

create-jobs-table! clj

(create-jobs-table! ds)

Create the job_queue table + ready-scan index if absent, and job_store with it, on H2, PostgreSQL or MySQL. Call once at startup, or ship as a migration.

Both, because enqueue-job! writes a history row: a caller who created only the queue table would fail on the next insert (BOU-418 review).

create-recurring-tables! clj

(create-recurring-tables! ds)

Create job_idempotency_keys, job_recurring and the ticker lease row if absent. Called at boot under :provider :db.

default-lease-ms clj

How long a claimed job may stay in-flight before reclaim-abandoned-jobs! considers its worker dead and returns it to the ready queue.

enqueue-in-tx! clj deprecated

(enqueue-in-tx! tx queue-name job)

DEPRECATED — use wagoe.jobs.ports/enqueue-in-tx! on the queue component.

Kept so existing callers keep working, but reaching into this adapter namespace couples application code to one implementation. The capability now lives on ports/ITransactionalJobQueue, so callers can stay on the port and ask ports/transactional-queue? instead of knowing which adapter they have.