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

#include <cwist/sys/job/durable_queue.h>
#include <cwist/core/mem/alloc.h>
#include <cwist/net/redis/cwist_redis.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

Include dependency graph for durable_queue.c:

Go to the source code of this file.

Macros
#define	_POSIX_C_SOURCE 200809L

#define	CWIST_JQ_GROUP "cwist-jq"

#define	CWIST_JQ_DEFAULT_VISIBILITY_MS 30000u

#define	CWIST_JQ_DEFAULT_MAX_RETRIES 5u

#define	CWIST_JQ_MAX_POLL_SLICE_MS 200u /* max single XREADGROUP BLOCK */

Typedefs
typedef cwist_error_t(*	jq_dead_count_fn) (cwist_job_queue_t q, uint64_t out_count)

Functions
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_job_queue_t *	cwist_job_queue_create_redis (cwist_redis_t conn, const cwist_job_queue_opts_t opts)
	Create a durable job queue backed by Redis streams.

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)
	Enqueue a job payload with an optional type.

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)
	Negatively acknowledge a claimed job (retry or dead-letter) and free it.

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 job queue backends over Redis (streams) and NATS (JetStream).

EXPERIMENTAL (v3.7 Phase 4). See include/cwist/sys/job/durable_queue.h.

Redis layout (prefix cwist:jq:<name>:):

<prefix>s stream; entries carry type + payload fields
<prefix>a hash: entry id -> delivery count (visibility-timeout redeliveries increment it; ack/nack/dead-letter remove it)
<prefix>delay zset: nacked jobs parked until their due time; the blob is [type_len:4BE][attempts:4BE][type][payload]
<prefix>dead list: dead-lettered blobs

Claim order per poll: reap due delayed jobs (Lua, atomic), reclaim stale pending entries via XAUTOCLAIM (visibility timeout), then read new entries via XREADGROUP. Dead-lettering happens when the delivery count exceeds max_retries.

NATS layout: JetStream stream <name> (subject <name>.jobs) with a durable pull consumer (AckWait = visibility, MaxDeliver = max_retries+1 as headroom). Exhausted deliveries are terminated and republished to a companion <name>_DEAD stream, so both backends expose a dead count.

Definition in file durable_queue.c.

Macro Definition Documentation

◆ _POSIX_C_SOURCE

#define _POSIX_C_SOURCE 200809L

Definition at line 25 of file durable_queue.c.

◆ CWIST_JQ_DEFAULT_MAX_RETRIES

#define CWIST_JQ_DEFAULT_MAX_RETRIES 5u

Definition at line 43 of file durable_queue.c.

◆ CWIST_JQ_DEFAULT_VISIBILITY_MS

#define CWIST_JQ_DEFAULT_VISIBILITY_MS 30000u

Definition at line 42 of file durable_queue.c.

◆ CWIST_JQ_GROUP

#define CWIST_JQ_GROUP "cwist-jq"

Definition at line 41 of file durable_queue.c.

◆ CWIST_JQ_MAX_POLL_SLICE_MS

#define CWIST_JQ_MAX_POLL_SLICE_MS 200u /* max single XREADGROUP BLOCK */

Definition at line 44 of file durable_queue.c.

Typedef Documentation

◆ jq_dead_count_fn

typedef cwist_error_t(* jq_dead_count_fn) (cwist_job_queue_t *q, uint64_t *out_count)

Definition at line 85 of file durable_queue.c.

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.

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.

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 job queue backed by Redis streams.

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.

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
	)

Enqueue a job payload with an optional type.

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
	)

Negatively acknowledge a claimed job (retry or dead-letter) and free it.

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.

Macros

Typedefs

Functions

Detailed Description

Macro Definition Documentation

◆ _POSIX_C_SOURCE

◆ CWIST_JQ_DEFAULT_MAX_RETRIES

◆ CWIST_JQ_DEFAULT_VISIBILITY_MS

◆ CWIST_JQ_GROUP

◆ CWIST_JQ_MAX_POLL_SLICE_MS

Typedef Documentation

◆ jq_dead_count_fn

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()