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

Consumer files are loaded before the Queue plugin activates in development, source-based serving, and compiled production bundles. Compiled starts use the bundled app-module registry, so the declared consumers are registered without a runtime TypeScript import.

// 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.

The Queue panel chooses the configured instance and its broker addresses from the plugin inventory. Its consumer registry and topic counters cover this process, including all queue instances in it. They are not fleet totals. A successful inspect read alone is not evidence of a live broker connection.

producerSelected identifies whether this instance owns the app-wide provider slot used by the outbox bridge and CDC sink. It is not a broker-health probe.

The Queue panel separates consumer policies from topic activity: group, maximum attempts and dead-letter target are visible even before a failure. Search matches topics, groups and dead-letter targets. Shared process-wide counters appear once per topic rather than once per consumer group. They accumulate since process start and do not measure pending messages or broker lag. Kafka dead-letter recovery remains a broker operation: inspect the recorded error, fix the handler or payload and replay through your broker tooling.

Production counters include successful QueueService.produce batches and outbox deliveries. A batch increments the counter only after the transport resolves; rejected batches do not increment it.

#Literal topic names in dashboard counters

The queue dashboard reads process-local counters keyed by the literal topic name. Names such as constructor, toString, and __proto__ are counted normally, including retries, dead letters and produced messages. A topic without activity has zero counters; it never inherits JavaScript object properties. These counters describe the selected replica, not broker-wide consumer-group totals.

The dashboard search filters consumer policies, topic activity and produced-message counts together. A consumer-group or dead-letter-topic match also includes production for the matching source topic. Topics with only produced messages remain searchable by name. Empty matches are distinguished from no recorded production; Clear search restores the full view.

The Queue dashboard keeps loading, empty configuration and failed observations separate. Invalid snapshots hide retained consumer rows and topic counters instead of being rendered as current activity. A read failure is shown once; Load again retries when the dashboard transport supports refresh.

On narrow screens, the topic-activity and produced-message tables scroll horizontally within their own regions. Focus either named table region with the keyboard and use the arrow keys to reach its columns. Long topic names and error details wrap without stretching the page.