Consuming Jobs

A producer can request a report without knowing how to build it. The worker supplies that implementation by registering a handler on a *queue.Consumer. The consumer finds the handler by job name, decodes the payload, executes it, and asks the driver to settle the delivery according to the result.

Start with a configured driver and the shared BuildReportJob and BuildReport payload from Producing Jobs. The consumer owns execution policy, not the driver’s connection lifecycle.

Registering Handlers

Register handlers before starting the worker. With context imported, queue from arkoren.dev/core/service/queue, and your application’s reports.Build(ctx, reportID) error operation, connect the job to its implementation:

consumer := queue.NewConsumer(driver)
err := consumer.Handle(BuildReportJob, func(ctx context.Context, payload BuildReport) error {
	return reports.Build(ctx, payload.ReportID)
})

Each attempt gets a fresh decoded payload. The context carries consumer cancellation, the job’s execution deadline when configured, and propagated trace context when telemetry is enabled. Pass it to database, storage, and HTTP calls. A timeout cancels the context; it does not stop application code by force.

Handle err before running the worker. Invalid definitions return queue.ErrInvalidName wrapping the definition error. A second registration for the same job name returns queue.ErrDuplicateName, even if its payload type differs. A nil handler is rejected without registration.

Run discovers queues once when it starts. A later handler on an already consumed queue can handle subsequent deliveries, but adding a new queue does not start another consume call until a later run. Register everything first to avoid timing-dependent coverage.

Selecting Queues

By default, a consumer reads every queue that has a registered handler. If report generation needs dedicated workers, restrict an alternative consumer construction:

consumer := queue.NewConsumer(driver, func(worker *queue.ConsumerBuilder) {
	worker.Queues("reports")
})

Register handlers on that consumer as above. Other handlers may remain registered, but their queues are skipped. This restriction neither changes driver routing nor provisions broker resources. A selected queue without a handler is not consumed; if none remain, Run returns queue.ErrNoQueues.

Each Queues call replaces the previous restriction; Queues() restores all handled queues. Inputs and the published restriction are copied. Construction callbacks run once in order, and a retained draft cannot reconfigure the published consumer. Telemetry and event buses remain shared services.

Worker Lifetime

Handle does not begin consumption. Register Run as a supervised runner in an existing application composition callback, with application imported from arkoren.dev/core/application:

compose := func(composer *application.Composer) {
	composer.Run(consumer.Run)
}

Pass the callback to your existing app.Run(ctx, compose). Include the driver’s startup, readiness, and cleanup hooks from Drivers when required. RabbitMQ and NATS need those hooks; memory needs cleanup but no readiness probe; SQS and Cloudflare have no driver hooks. Construction alone does not connect RabbitMQ or NATS.

Run starts one driver Consume call per selected, handled queue. Shipped drivers execute handlers sequentially within each call, including fetched batches. Different queues run concurrently. For more concurrency on one queue, run additional workers sharing the same physical work queue or JetStream durable; they compete for deliveries rather than receiving a broadcast copy each. Prefetch and batch size reserve work ahead, not parallel handler slots.

If any consume call returns an error, Run cancels its sibling queue consumers, waits for them, and returns the first error. As an application runner, that error stops the service; let your deployment supervisor restart a worker after broker failure rather than silently leaving it idle. Ordinary handler errors whose retry or failure settlement succeeds do not themselves end Run.

Stopping Workers

Cancel the consumer and wait for it to return before stopping its driver or closing a borrowed connection. The active handler must cooperate with cancellation; the consumer cannot kill it. A handler that completes successfully still needs its acknowledgement, so allow shutdown time for execution and settlement.

Settlement has a fresh 30-second budget after execution and is detached from consumer cancellation. It can still fail if transport cleanup happens first or the backend is unavailable. If the application’s shutdown budget expires, cleanup can release a driver while a handler is still running. See shutdown settlement for the retry exception and its attempt-count implications.

Memory consumer cancellation preserves work not yet admitted; it does not drain the backlog or revoke the active delivery. An active handler can acknowledge or retry with the detached settlement context. memoryqueue.Stop is different: it discards ready jobs and delayed retries, prevents new admissions, and waits for active work only until its context ends. Retrying into that stopped driver fails. Use a broker-backed driver if pending work must outlive the process.

Handler Results

Return nil only when the work is complete; swallowing an error requests acknowledgement of unfinished work. Return an ordinary error for a transient failure. The consumer requests retry using the registered job’s attempt limit and backoff, or terminal failure if attempts are exhausted.

Use queue.Permanent(err) when another attempt cannot help. queue.RetryAfter requests a specific delay while still respecting the attempt cap. queue.Reject delegates unwanted-message handling to backend rejection policy; it is not a portable delete or dead-letter operation. Choose among these using Delivery and Retries.

Make reports.Build repeat-safe with the stable report ID. A successful report write followed by a lost acknowledgement can run again, and a lost lease can overlap another attempt. For database-only effects, record completion and the effect in one transaction. For remote effects, use provider idempotency and reconcile uncertain results; a local completion flag cannot make a remote call atomic. Required ordering also belongs in application state.

Job timeout and broker reservation are separate. NATS and SQS try to renew reservations, but renewal failure can permit duplicates. Cloudflare cannot renew its lease: all earlier handlers in a fetched batch consume the same lease time, so keep batches small and finish execution plus settlement before expiry.

Testing Handlers

Test the report-building operation directly for its application behavior. To exercise JSON decoding and consumer settlement too, use arkoren.dev/core/service/queue/queuetest. In a test with t *testing.T, construct both facades around the same fake:

fake := queuetest.New()
dispatcher := queue.NewDispatcher(fake)
consumer := queue.NewConsumer(fake)

Register your handler using Handle and check its registration error. Dispatch a report payload, then drain due work through that consumer:

if err := dispatcher.Dispatch(t.Context(), BuildReportJob, BuildReport{ReportID: "rpt-204"}); err != nil {
	t.Fatal(err)
}
if err := fake.Drain(t.Context(), consumer); err != nil {
	t.Fatal(err)
}

Assert the report’s observable stored result using your application’s test fixtures. Successful draining proves pending due work was processed without a settlement error, not that every handler succeeded: a terminally failed job also drains. Inspect the fake’s Failed() or Rejected() records when testing those outcomes.

Delayed retries remain pending by default. queuetest.WithScheduled() includes them by skipping waits, not advancing time; use it when testing attempt policy, not elapsed-time behavior. A per-message delivery limit catches endless retry loops. See running dispatched jobs for assertions and fake options.

See the queue API and queuetest API.