Durable Job Queue

Durable Job Queue

Header: <cwist/sys/job/durable_queue.h>

A persistent job queue with at-least-once delivery, backed by Redis Streams or NATS JetStream. Jobs survive process restarts, unacknowledged jobs are redelivered after a visibility timeout, and jobs that keep failing are dead-lettered. Supported since v3.8; nothing is enabled unless you create a queue. For in-process background work without persistence, use the Scheduler.

#include <cwist/sys/job/durable_queue.h>

cwist_redis_t *r = cwist_redis_connect("127.0.0.1", 6379);
cwist_job_queue_opts_t opts = {
    .name = "emails",
    .visibility_timeout_ms = 30000,
    .max_retries = 5,
};
cwist_job_queue_t *q = cwist_job_queue_create_redis(r, &opts);

cwist_job_queue_enqueue(q, "{\"to\":\"a@b.co\"}", 15, "welcome");

/* worker loop */
cwist_job_t *job = NULL;
cwist_error_t err = cwist_job_queue_claim(q, 5000, &job);
if (cwist_error_is_ok(&err)) {
    size_t len;
    const void *payload = cwist_job_payload(job, &len);
    bool ok = send_email(payload, len, cwist_job_type(job));
    if (ok) cwist_job_queue_ack(q, job);
    else    cwist_job_queue_nack(q, job);
}

cwist_job_queue_destroy(q);
cwist_redis_close(r);

Backends

	Redis	NATS JetStream
Create with	cwist_job_queue_create_redis(conn, &opts)	cwist_job_queue_create_nats(nats, &opts)
Storage	one stream per queue with a consumer group (cwist:jq:<name>:*), created on first use	one stream per queue with a durable pull consumer
Claiming	XREADGROUP; expired jobs reclaimed with XAUTOCLAIM	pull consumer, AckWait = visibility timeout
Dead letters	list <prefix>:dead	stream <name>_DEAD
Requires	Redis 6.2+ for XAUTOCLAIM	a JetStream-enabled server

The queue does not own the connection; it must outlive the queue.

Options

Field	Default	Meaning
name	required	queue name (Redis key prefix, JetStream stream name)
consumer	unique per worker	consumer identity in the group
visibility_timeout_ms	30000	an unacknowledged job becomes claimable again after this
max_retries	5	deliveries (including the first) before dead-lettering; at least 1
retry_delay_ms	0	delay before a nacked job is claimable again

API

Function	Description
cwist_error_t cwist_job_queue_enqueue(q, payload, len, type)	Persist a job (binary-safe payload, optional type tag).
cwist_error_t cwist_job_queue_claim(q, timeout_ms, &job)	Claim one job, waiting up to timeout_ms (0 polls once). Returns CWIST_ERROR_TIMEOUT when none arrives.
cwist_error_t cwist_job_queue_ack(q, job)	Remove the job permanently.
cwist_error_t cwist_job_queue_nack(q, job)	Return it for a retry, or dead-letter it once retries are exhausted.
void cwist_job_destroy(cwist_job_t *job)	Drop the handle without ack or nack; the job is reclaimed after the visibility timeout.
cwist_job_payload(job, &len), cwist_job_type(job), cwist_job_id(job), cwist_job_attempts(job)	Job data, type, backend id (stream entry id or sequence) and delivery count.
cwist_error_t cwist_job_queue_dead_count(q, &count)	Jobs in the dead-letter store.
void cwist_job_queue_destroy(q)	Free the handle; claimed jobs are reclaimed by other workers later.

Because delivery is at-least-once, job handlers must be idempotent.

CI gate

CI runs the queue tests against pinned Redis and NATS JetStream servers and fails if a backend is skipped; see Durable Queue CI Gate.