Worker

import "github.com/gp-system/gpsystem/worker"

worker.Run is the background-work twin of the HTTP engines' Run: the same app lifecycle (signal handling, graceful shutdown, closers, telemetry flush), except that instead of an HTTP server it runs three things in one binary:

• the asynq server: processing the event, listener, scheduled and job tasks;
• the outbox relay: forwarding committed events to the queue;
• the schedule runner: firing the cron entries, with leader election.

This runs task processing and scheduling in a single process: instead of separate worker and scheduler deployments, you scale one cmd/worker binary. The binary comes with new project (for older projects add worker backfills it), and you deploy and scale it independently of the API.

The generated cmd/worker

The shop's worker binary: every line written by the generator; modules are wired in by new module / add listener / add job at the // gpsystem:* anchors.

package main

import (
    "context"
    "log"

    "github.com/gp-system/gpsystem/dbx/pg"
    "github.com/gp-system/gpsystem/envconf"
    "github.com/gp-system/gpsystem/events"
    "github.com/gp-system/gpsystem/events/outbox"
    "github.com/gp-system/gpsystem/scheduler"
    "github.com/gp-system/gpsystem/worker"

    "github.com/acme/shop/internal/modules/shop"
    shopjobs "github.com/acme/shop/internal/modules/shop/jobs"
    shoplisteners "github.com/acme/shop/internal/modules/shop/listeners"
    "github.com/acme/shop/internal/platform/config"
)

func main() {
    cfg := envconf.MustLoad[config.Config]()
    ctx := context.Background()
    pool := pg.MustNewPool(ctx, cfg.DB)
    relay := outbox.MustNewRelay(ctx, cfg.Outbox, pool, cfg.Worker.Redis)

    err := worker.Run(ctx, cfg.Worker, func(reg *events.Registry, sched *scheduler.Schedule) error {
        shopDeps := shop.Dependencies{
            DB:         pg.NewDB(pool),
            Transactor: pg.NewTransactor(pool),
            Dispatcher: outbox.NewDispatcher(outbox.NewStore(pg.NewDB(pool))),
        }
        shoplisteners.Register(reg, shopDeps)
        shopjobs.Register(sched, shopDeps)
        return nil
    },
        worker.WithOutboxRelay(relay),
        worker.WithCloser("pgxpool", func(context.Context) error {
            pool.Close()
            return nil
        }),
    )
    if err != nil {
        log.Fatal(err)
    }
}

The pattern is the same as on the HTTP side: dependencies travel in an explicit Dependencies struct, and the register callback receives the event Registry and the Schedule, filled by the Register functions of the modules' generated listeners / jobs subpackages. The worker's Dispatcher is the outbox one too: an event dispatched by a listener gets the same guarantee as one dispatched from an HTTP request.

Run

func Run(ctx context.Context, cfg Config, register RegisterFunc, opts ...Option) error

type RegisterFunc func(reg *events.Registry, sched *scheduler.Schedule) error

Run, in order: telemetry bootstrap (telemetry.Setup) → register builds the registry and the schedule → opens a queue.Client (PING) → starts the asynq server, alongside the relay (when you passed WithOutboxRelay) and the schedule runner (when the schedule is non-empty; an empty schedule means no leader election either) → blocks until SIGINT/SIGTERM. Returns nil on a clean shutdown.

What processes the tasks

The worker registers four built-in handlers on the task-type prefixes:

Every handler runs through three middleware: a per-task Sentry hub (breadcrumb isolation), panic recovery (a panicking handler becomes an errs error and retries instead of killing the worker), and tracing: the consumer span continues the envelope's trace context, so the whole path is one trace. A task with no registered handler (say, a deploy removed the listener) is skipped with a warning rather than retried forever.

Shutdown order

On SIGINT/SIGTERM, Run shuts down in a deterministic order:

1. the producing loops stop: the schedule runner (releasing its lease) and the relay's poll loop;
2. in-flight tasks drain: for at most WORKER_SHUTDOWN_TIMEOUT; whatever doesn't finish is requeued by asynq and redelivered later (one more reason for idempotent listeners);
3. closers run LIFO: first the kit's own (scheduler, relay, queue client), then your WithClosers (in the shop, the pgx pool);
4. telemetry flush: the last spans and logs still go out.

It is the same guarantee as on the HTTP side: a deploy doesn't cut an e-mail send in half. See the application lifecycle for details.

Options

The two escape hatches are the worker-side half of the kit's "centralize, don't abstract" principle: when you need an asynq capability, you don't have to leave the chassis.

worker.Run(ctx, cfg.Worker, register,
    worker.WithAsynqConfig(func(c *asynq.Config) {
        c.HealthCheckFunc = reportHealth
    }),
    worker.WithMux(func(mux *asynq.ServeMux) {
        mux.HandleFunc("import:products", handleProductImport) // your own task type
    }),
)

Queue priorities

The worker serves queues by weight: WORKER_QUEUES is a name:weight list, the counterpart of Horizon's balance/queue weighting. The shop puts its mail on a separate mail queue (the listener's queue.OnQueue("mail") registration), so a mail burst can't crowd out the rest of the work:

WORKER_QUEUES=default:3,mail:1

With that, the worker spends ~75% of its capacity on default and ~25% on mail whenever both have work. WORKER_STRICT_PRIORITY=true replaces weighting with a strict order: the highest-weight queue is drained completely before the next one is touched, useful when you have a critical queue that trumps everything.

Configuration

You embed worker.Config without a prefix in the project config (ready-made in the generated config.go):

type Config struct {
    Worker worker.Config
    DB     dbx.Config    `envPrefix:"DB_"`
    Outbox outbox.Config `envPrefix:"OUTBOX_"`
}

worker.Config embeds three more configs, so the worker binary is configured from a single set of env vars:

• queue.Config under the REDIS_ prefix (REDIS_ADDR, ...),
• scheduler.Config under the SCHEDULER_ prefix,
• telemetry.Config without a prefix (the standard OTEL_*, LOG_*, SENTRY_* names); see the observability overview.

The full env reference: configuration.

Logging

The worker logs per task through the default slog logger, trace-correlated with the consumer span:

• worker: task started (Info): with task_type, task_id, attempt attributes;
• worker: task done (Info): plus duration;
• worker: task failed (Error): plus error; an intentional skip (SkipRetry, e.g. no registered handler) appears only once, as a warning.

asynq's own internal logs flow into the same pipeline. Where the lines end up (dev console, JSON stdout, OTLP) is decided by the logging config; LOG_LEVEL=WARN silences the per-task Info lines.

Scaling

The worker is stateless: run as many replicas as your workload needs, and all three components are replica-safe.

• the asynq server by nature: one task goes to one worker at a time;
• the outbox relay via the FOR UPDATE SKIP LOCKED + deterministic TaskID pair;
• the schedule runner via the Redis-lease leader election: one of N replicas fires.

Nothing needs to be switched off or deployed separately to scale. The price is a single contract: delivery is at-least-once, so listeners are idempotent.

Running locally

go run ./cmd/migrate up   # also applies the outbox_events migration
mise run worker           # go run ./cmd/worker

Redis comes from the project's compose.yml (mise run dev). For the generator commands (add event, add listener, add job, add worker), see the CLI worker generators.