Durable (persistent) job queue backends over Redis and NATS. More...

#include <cwist/sys/err/cwist_err.h>
#include <stddef.h>
#include <stdint.h>

Include dependency graph for durable_queue.h:

This graph shows which files directly or indirectly include this file:

Classes
struct	cwist_job_queue_opts

Typedefs
typedef struct cwist_redis	cwist_redis_t

typedef struct cwist_nats	cwist_nats_t

typedef struct cwist_job_queue	cwist_job_queue_t

typedef struct cwist_job	cwist_job_t

typedef struct cwist_job_queue_opts	cwist_job_queue_opts_t

Functions
cwist_job_queue_t *	cwist_job_queue_create_redis (cwist_redis_t conn, const cwist_job_queue_opts_t opts)
	Create a durable queue over an existing Redis connection.

cwist_job_queue_t *	cwist_job_queue_create_nats (cwist_nats_t nats, const cwist_job_queue_opts_t opts)
	Create a durable queue over an existing NATS connection.

void	cwist_job_queue_destroy (cwist_job_queue_t *q)
	Destroy a queue and all backend resources it owns; NULL is a no-op. In-flight jobs are not touched.

cwist_error_t	cwist_job_queue_enqueue (cwist_job_queue_t q, const void payload, size_t payload_len, const char *type)
	Persist a job for later processing.

cwist_error_t	cwist_job_queue_claim (cwist_job_queue_t q, uint64_t timeout_ms, cwist_job_t *job)
	Claim one job, waiting up to `timeout_ms`.

cwist_error_t	cwist_job_queue_ack (cwist_job_queue_t q, cwist_job_t job)
	Acknowledge a claimed job and free it.

cwist_error_t	cwist_job_queue_nack (cwist_job_queue_t q, cwist_job_t job)
	Return a claimed job to the queue.

void	cwist_job_destroy (cwist_job_t *job)
	Public alias of cwist_job_free(); releases a job returned by claim.

const void *	cwist_job_payload (const cwist_job_t job, size_t len)
	Return the job payload pointer and store its length in len (0 on NULL job).

const char *	cwist_job_type (const cwist_job_t *job)
	Return the job type string, or NULL if none / NULL job.

const char *	cwist_job_id (const cwist_job_t *job)
	Return the backend entry id of the job, or NULL for a NULL job.

uint32_t	cwist_job_attempts (const cwist_job_t *job)
	Return the delivery attempt count of the job (0 for a NULL job).

cwist_error_t	cwist_job_queue_dead_count (cwist_job_queue_t q, uint64_t out_count)
	Store the number of dead-lettered jobs in *out_count.

Detailed Description

Durable (persistent) job queue backends over Redis and NATS.

Supported since v3.8: opt-in tier on top of the existing Redis and NATS clients, separate from the in-process scheduler queue (cwist/sys/job/scheduler.h). Jobs survive process restarts; delivery is at-least-once.

Backends:

Redis: stream per queue with a consumer group. Claiming uses XREADGROUP; unacknowledged jobs are reclaimed via XAUTOCLAIM after the visibility timeout; nacks and exhausted retries are dead-lettered to a per-queue list (<prefix>:dead).
NATS: JetStream stream per queue with a durable pull consumer (AckWait = visibility timeout, MaxDeliver = max_retries + 1). Exhausted messages are terminated and republished to a companion <name>_DEAD stream.

Nothing here is enabled by default; an app opts in by creating a queue handle and submitting jobs to it.

Definition in file durable_queue.h.

Typedef Documentation

◆ cwist_job_queue_opts_t

typedef struct cwist_job_queue_opts cwist_job_queue_opts_t

Queue creation options.

◆ cwist_job_queue_t

typedef struct cwist_job_queue cwist_job_queue_t

Opaque durable queue handle.

Definition at line 39 of file durable_queue.h.

◆ cwist_job_t

typedef struct cwist_job cwist_job_t

A single claimed job. Valid until ack/nack/cwist_job_destroy.

Definition at line 42 of file durable_queue.h.

◆ cwist_nats_t

typedef struct cwist_nats cwist_nats_t

Definition at line 36 of file durable_queue.h.

◆ cwist_redis_t

typedef struct cwist_redis cwist_redis_t

Definition at line 35 of file durable_queue.h.

Function Documentation

◆ cwist_job_attempts()

uint32_t cwist_job_attempts ( const cwist_job_t * job )

Return the delivery attempt count of the job (0 for a NULL job).

Number of deliveries so far, including the current one.

Definition at line 216 of file durable_queue.c.

References cap.

◆ cwist_job_destroy()

void cwist_job_destroy ( cwist_job_t * job )

Public alias of cwist_job_free(); releases a job returned by claim.

Release a job handle without ack/nack (the job is reclaimed later). Always safe on NULL.

Definition at line 191 of file durable_queue.c.

References cap.

◆ cwist_job_id()

const char * cwist_job_id ( const cwist_job_t * job )

Return the backend entry id of the job, or NULL for a NULL job.

Backend delivery token (Redis stream entry id / NATS sequence).

Definition at line 211 of file durable_queue.c.

References cap.

◆ cwist_job_payload()

const void * cwist_job_payload	(	const cwist_job_t *	job,
		size_t *	len
	)

Return the job payload pointer and store its length in len (0 on NULL job).

Job payload bytes (binary-safe); *len receives the length.

Definition at line 196 of file durable_queue.c.

References cap.

◆ cwist_job_queue_ack()

cwist_error_t cwist_job_queue_ack	(	cwist_job_queue_t *	q,
		cwist_job_t *	job
	)

Acknowledge a claimed job and free it.

Permanently remove a claimed job.

Return values

CWIST_ERROR_INVALID_PARAM if the job does not belong to q.

Definition at line 990 of file durable_queue.c.

References cap, and CWIST_ERROR_INVALID_PARAM.

◆ cwist_job_queue_claim()

cwist_error_t cwist_job_queue_claim	(	cwist_job_queue_t *	q,
		uint64_t	timeout_ms,
		cwist_job_t **	job
	)

Claim one job, waiting up to timeout_ms.

The caller must eventually cwist_job_queue_ack(), cwist_job_queue_nack(), or cwist_job_destroy() the returned job; an unacked job is reclaimed after the visibility timeout.

Returns: 0 on success, CWIST_ERROR_TIMEOUT when no job appears within timeout_ms (0 = single poll).

Claim one job, waiting up to timeout_ms.

Returns: 0 with *job set (NULL on parameter error); backend errors propagate.

Definition at line 976 of file durable_queue.c.

References cap, and CWIST_ERROR_INVALID_PARAM.

◆ cwist_job_queue_create_nats()

cwist_job_queue_t * cwist_job_queue_create_nats	(	cwist_nats_t *	nats,
		const cwist_job_queue_opts_t *	opts
	)

Create a durable queue over an existing NATS connection.

Requires a JetStream-enabled server. Ensures the job stream and a durable pull consumer exist. Takes no ownership of nats; the connection must outlive the queue.

Definition at line 922 of file durable_queue.c.

References cap.

◆ cwist_job_queue_create_redis()

cwist_job_queue_t * cwist_job_queue_create_redis	(	cwist_redis_t *	conn,
		const cwist_job_queue_opts_t *	opts
	)

Create a durable queue over an existing Redis connection.

Ensures the consumer group exists (created on first use, MKSTREAM). Takes no ownership of conn; the connection must outlive the queue.

Create a durable queue over an existing Redis connection.

Parameters

conn	Borrowed Redis connection (must outlive the queue).

Returns: New queue with consumer group ensured, or NULL on invalid opts, ENOMEM, or Redis error. Frees nothing on conn.

Definition at line 591 of file durable_queue.c.

References cap, cwist_alloc(), cwist_free(), CWIST_JQ_GROUP, and cwist_strdup().

◆ cwist_job_queue_dead_count()

cwist_error_t cwist_job_queue_dead_count	(	cwist_job_queue_t *	q,
		uint64_t *	out_count
	)

Store the number of dead-lettered jobs in *out_count.

Number of jobs currently parked in the dead-letter store.

Return values

CWIST_ERROR_INVALID_PARAM on NULL arguments; CWIST_FAILURE if the backend has no dead-count implementation.

Definition at line 1021 of file durable_queue.c.

References cap, CWIST_ERROR_INVALID_PARAM, and CWIST_FAILURE.

◆ cwist_job_queue_destroy()

void cwist_job_queue_destroy ( cwist_job_queue_t * q )

Destroy a queue and all backend resources it owns; NULL is a no-op. In-flight jobs are not touched.

Destroy the queue handle. Claimed-but-unacked jobs are reclaimed by other workers once their visibility timeout expires.

Definition at line 933 of file durable_queue.c.

References cap, and cwist_free().

◆ cwist_job_queue_enqueue()

cwist_error_t cwist_job_queue_enqueue	(	cwist_job_queue_t *	q,
		const void *	payload,
		size_t	payload_len,
		const char *	type
	)

Persist a job for later processing.

Parameters

payload	Opaque payload bytes; binary-safe.
type	Optional type/tag string (NULL for none).

Persist a job for later processing.

Return values

CWIST_ERROR_INVALID_PARAM on NULL queue or NULL payload with nonzero length.

Definition at line 960 of file durable_queue.c.

References cap, and CWIST_ERROR_INVALID_PARAM.

◆ cwist_job_queue_nack()

cwist_error_t cwist_job_queue_nack	(	cwist_job_queue_t *	q,
		cwist_job_t *	job
	)

Return a claimed job to the queue.

The job becomes claimable again after retry_delay_ms; once deliveries exceed max_retries it is dead-lettered instead.

Return a claimed job to the queue.

Return values

CWIST_ERROR_INVALID_PARAM if the job does not belong to q.

Definition at line 1005 of file durable_queue.c.

References cap, and CWIST_ERROR_INVALID_PARAM.

◆ cwist_job_type()

const char * cwist_job_type ( const cwist_job_t * job )

Return the job type string, or NULL if none / NULL job.

Job type/tag, or NULL when the job was enqueued without one.

Definition at line 206 of file durable_queue.c.

References cap.

Classes

Typedefs

Functions

Detailed Description

Typedef Documentation

◆ cwist_job_queue_opts_t

◆ cwist_job_queue_t

◆ cwist_job_t

◆ cwist_nats_t

◆ cwist_redis_t

Function Documentation

◆ cwist_job_attempts()

◆ cwist_job_destroy()

◆ cwist_job_id()

◆ cwist_job_payload()

◆ cwist_job_queue_ack()

◆ cwist_job_queue_claim()

◆ cwist_job_queue_create_nats()

◆ cwist_job_queue_create_redis()

◆ cwist_job_queue_dead_count()

◆ cwist_job_queue_destroy()

◆ cwist_job_queue_enqueue()

◆ cwist_job_queue_nack()

◆ cwist_job_type()