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 the producer does not need to know its consumers.

Plans

Queues use the queue-free, queue-starter, queue-basic, queue-premium, and queue-enterprise plans, all with durable storage: fixed at 1 GB on queue-free and 5 GB included on the other plans, expandable with the excess billed. queue-free and queue-starter are single-instance; queue-basic, queue-premium, and queue-enterprise are highly available with one primary and two replicas. The Premium and Enterprise plans include instance metrics on the project page. Check prices and capacity in the Valkey catalog.

Required permissions

Creating a queue requires managed_service.create on organization:*. To read status, use managed_service.status.read on managed_service:<project-id>; for the masked connection, use managed_service.connection.read; for rotation, use managed_service.credentials.rotate. owner has full access; assistant, member, and organization API Keys need the corresponding grants.

Recommended Streams flow

1. A producer adds a task with XADD.
2. The group is created once with XGROUP CREATE ... MKSTREAM.
3. A worker reads new messages with XREADGROUP.
4. The worker processes the task and only then acknowledges it with XACK.
5. A recovery process inspects XPENDING and uses XAUTOCLAIM to take over messages from inactive consumers.
6. 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

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.

Next steps

• Connect to Valkey
• Key‑Value
• Cache
• Valkey Queue benchmarks
• Managed services API