Queue (Kafka interop)

@voltro/plugin-queue is the door to queues somebody else owns: a Voltro backend consuming and producing against an adopter's existing Kafka. The boundary with the built-ins, in one line each: the outbox is your own durable side-effects, a workflow is your own orchestration — this plugin is interop with foreign infrastructure. Kafka first; the provider contract is cut so SQS/RabbitMQ can be later implementations.

// app.config.ts
import { queuePlugin } from '@voltro/plugin-queue'

export default {
  // …
  plugins: [
    queuePlugin({ brokers: ['kafka-1:9092', 'kafka-2:9092'] }),
  ],
}

#Consuming: *.consumer.ts

// src/consumers/orders.consumer.ts
import { Schema } from 'effect'
import { defineQueueConsumer } from '@voltro/plugin-queue'

export const orders = defineQueueConsumer({
  topic: 'orders',
  schema: Schema.Struct({ orderId: Schema.String, total: Schema.Number }),
  handler: async (order, ctx) => {
    // MUST be idempotent — delivery is at-least-once. A unique-column
    // upsert is the standard shape:
    await ctx.store.upsert('orders_mirror',
      { orderId: order.orderId, total: order.total },
      { conflictColumns: ['orderId'] })
  },
})

Both boot paths discover *.consumer.ts; the plugin starts every registered consumer at activation and stops them at shutdown. The semantics, precisely:

• Ordering: serial per partition. Parallelism exists only ACROSS partitions — concurrency inside one would destroy ordering and commit semantics both. Retry backoff deliberately BLOCKS the partition.
• Commit after the handler, per message. A process killed mid-batch redelivers exactly the unhandled tail — never the whole batch, never a skipped message.
• Decode failures dead-letter IMMEDIATELY (to <topic>.dlq, with x-voltro-dlq-* reason headers) — a deterministic failure retried forever is an infinite loop with extra steps, and a poison message must release its partition.
• Handler failures retry with backoff, then dead-letter after maxAttempts (default 3).
• A rebalance is not a failure. A partition revoked mid-batch or mid-retry stops processing without a retry-counter increment or a DLQ publish — the new owner redelivers.
• Handlers are NOT wrapped in a transaction (the same documented boundary as HTTP route handlers). A handler needing atomic multi-writes opens ctx.store.transactional itself — and stays idempotent either way.
• Replica coordination is Kafka's own. Every replica joins the same consumer group and the broker assigns partitions — no advisory lock, unlike schedules, which coordinate through the claim table because no broker exists to do it for them.

#Producing

Two paths, one rule: transactional-with-a-write goes through the outbox.

// Inside a mutation — commits or rolls back WITH the domain write:
await ctx.outbox.enqueue('queue.produce', {
  topic: 'orders',
  messages: [{ key: order.id, value: JSON.stringify(order) }],
})

// src/queue.outbox.ts — the bridge (once per app):
import { queueOutboxHandler } from '@voltro/plugin-queue'
export default queueOutboxHandler()

The outbox runner delivers after commit — batched (messages is an array → one transport round-trip), at-least-once, retried, dead-lettered. One durability path: the existing outbox, not a second one. For fire-and-forget producing without a surrounding write, yield* QueueService in a handler and call produce(topic, messages) directly.

Topic creation is EXPLICIT (provider.ensureTopics([...])) — your Kafka is foreign infrastructure, and whether a client may create topics is your policy. A consumer started against a topic that does not exist yet warns and retries in the background (it connects once the topic appears), never aborting the boot.

#cdc-out to Kafka

kafkaSink plugs table-change mirroring (plugin-cdc-out) into the SAME provider: message key = the row id (one row's changes stay ordered in one partition), value = the change record, and the x-voltro-delivery-key header carries cdc-out's at-least-once dedupe handle.

import { cdcOutPlugin } from '@voltro/plugin-cdc-out'
import { kafkaSink } from '@voltro/plugin-queue'

cdcOutPlugin({ sinks: [{ table: 'orders', sink: kafkaSink({ topic: 'orders.cdc' }) }] })

#Observability

Metrics. Every counter is exported to the metrics registry, so it is scrapeable via @voltro/plugin-prometheus and readable at GET /_voltro/inspect/metrics:

Series	Type	What it answers
voltro_queue_consumed_total{topic,outcome}	counter	Throughput, and — with outcome = ok | dead-lettered — the dead-letter rate as a plain division.
voltro_queue_retries_total{topic}	counter	In-process handler retries. A retry BLOCKS its partition, so a rising rate is head-of-line latency, not just noise.
voltro_queue_produced_total{topic}	counter	Messages produced through the outbox bridge.
voltro_queue_lag_messages{topic,partition}	gauge	Backlog behind the message just picked up — "are the consumers keeping up", which no counter can answer.

outcome has two values on purpose: ok + dead-lettered is every message the runner finished with. A message abandoned by a rebalance is in neither — it was not consumed here, its new owner redelivers it and counts it there, and counting it twice would make the dead-letter ratio wrong in the direction of looking healthy.

Lag is a sample at pickup and costs nothing to collect (highWatermark rides along in the fetch response — no admin round trip per message). Read it together with the consume rate: nothing arrives to move the gauge on an idle or revoked partition, so it holds its last value, and a frozen high lag and a frozen low lag look identical on their own.

The per-topic counters plus the last error also stay on GET /_voltro/inspect/plugins/queue/consumers and in the dashboards' Queue panel — that view is this replica, right now, and carries an error string, which is not a time series. Both are moved by one recorder each, so they cannot drift.

Tracing. Each consumed message is processed inside a queue.consume span that ADOPTS the producer's traceparent header as its parent, so a Kafka hop no longer ends the trace. The span covers the whole message — decode, every retry, and the dead-letter publish — and carries messaging.system, messaging.destination.name, messaging.consumer.group.name, messaging.destination.partition.id, messaging.message.offset and voltro.queue.outcome (ok | dead-lettered | stale). A missing or malformed traceparent starts a fresh root span rather than failing the message. ctx.traceparent is still handed to your handler for hops the framework does not make for you.

Consumer spans are emitted from detached work — a broker callback, outside the server's Effect scope — and reach the server's tracer because the server publishes its tracer instance for exactly that case. There is still only ONE tracer: a second provider would mean a second exporter nothing flushes at shutdown. The same applies to cdcOut.deliver and plugin.<name>.schedule-fire, which run detached for the same reason.