Managed Queues

Zenifra's Queues component currently uses Valkey and is intended for asynchronous tasks and workers. The recommended pattern is Streams with consumer groups: messages remain in the stream, each worker receives part of the work, and producers do not need to know their consumers.

Recommended Streams flow

A producer adds a task with XADD.
The group is created once with XGROUP CREATE ... MKSTREAM.
A worker reads new messages with XREADGROUP.
The worker processes the task and only then acknowledges it with XACK.
A recovery process inspects XPENDING and uses XAUTOCLAIM to take over messages from inactive consumers.
The stream is bounded with approximate MAXLEN or XTRIM to prevent unbounded growth.

Streams normally provide at-least-once delivery. Processing must be idempotent: a task can be delivered again after a timeout, failure, or consumer recovery.

Node.js and TypeScript

npm install iovalkey

import Redis from 'iovalkey'
import { randomUUID } from 'node:crypto'

const url = new URL(process.env.VALKEY_URL ?? '')
const client = new Redis({
  host: url.hostname,
  port: Number(url.port),
  username: decodeURIComponent(url.username),
  password: decodeURIComponent(url.password),
  tls: { servername: url.hostname },
})

const stream = 'jobs:emails'
const group = 'email-workers'
const consumer = `worker-${randomUUID()}`

async function processEmail(fields) {
  console.log('process message', fields)
}

try {
  await client.xgroup('CREATE', stream, group, '$', 'MKSTREAM')
} catch (error) {
  if (!String(error).includes('BUSYGROUP')) throw error
}

await client.xadd(stream, 'MAXLEN', '~', 10000, '*', 'type', 'welcome', 'user_id', '42')
const batches = await client.xreadgroup('GROUP', group, consumer, 'COUNT', 10, 'BLOCK', 5000, 'STREAMS', stream, '>')

for (const [, messages] of batches ?? []) {
  for (const [id, fields] of messages) {
    await processEmail(fields)
    await client.xack(stream, group, id)
  }
}

await client.xtrim(stream, 'MAXLEN', '~', 10000)
await client.quit()

processEmail() represents the application work and must be idempotent. A separate process should inspect XPENDING and run XAUTOCLAIM when a consumer becomes inactive.

Python

python -m pip install valkey

import os
import uuid
from urllib.parse import urlparse
from valkey import Valkey
from valkey.exceptions import ResponseError

def process_email(fields):
    print("process message", fields)

url = urlparse(os.environ["VALKEY_URL"])
client = Valkey(
    host=url.hostname, port=url.port, username=url.username, password=url.password,
    ssl=True, ssl_check_hostname=True, decode_responses=True,
)

stream = "jobs:emails"
group = "email-workers"
consumer = f"worker-{uuid.uuid4()}"

try:
    client.xgroup_create(stream, group, id="$", mkstream=True)
except ResponseError as error:
    if "BUSYGROUP" not in str(error):
        raise

client.xadd(stream, {"type": "welcome", "user_id": "42"}, maxlen=10000, approximate=True)
for _, messages in client.xreadgroup(group, consumer, {stream: ">"}, count=10, block=5000):
    for message_id, fields in messages:
        process_email(fields)
        client.xack(stream, group, message_id)

client.xtrim(stream, maxlen=10000, approximate=True)
client.close()

process_email() represents the application work and must be idempotent. A separate process should inspect XPENDING and use XAUTOCLAIM to recover messages from inactive consumers.

Pub/Sub, Lists, and Streams

Feature	Best use	Main limitation
Streams	durable tasks, multiple workers, and recovery	requires consumer groups, `XACK`, retention, and idempotency
Lists	simple queue with one consumer per item	`BLPOP` removes before processing; prefer `BLMOVE` when loss matters
Pub/Sub	live notifications and ephemeral fan-out	messages are not persisted; consumer disconnects lose events

Pub/Sub is supported by the service, but it is not the primary implementation of the Queues profile. Use PUBLISH/SUBSCRIBE for presence, live invalidation, or notifications that may be lost. For business tasks, use Streams.

Limits and expectations

There is no exactly-once guarantee; use an application idempotency key.
Define retry, expiration, pending recovery, and a dead-letter strategy using another stream when needed.
Bound stream size with MAXLEN/XTRIM and monitor pending messages.
Do not treat plan high availability as confirmation of every message during every failure.
The product does not provide RabbitMQ/Kafka exchanges, routing keys, a native dead-letter queue, or automatic partitioning.