Queues and tasks

A task asks a worker to do one operation later. The producer submits it to Redis; a worker selects its handler and processes it outside the original request. Use a queue for work that needs retries, delayed execution, or survival across application restarts.

flowchart TB
    accTitle: One task, one worker per delivery
    accDescr: Producers enqueue into Redis. A worker runs the matching handler. A task may complete, retry after a delay, or be archived; another delivery can repeat the operation.
    Producer[HTTP, CLI or scheduler] -->|Enqueue| Redis[(Redis queue)]
    Redis -->|One competing worker| Handler[ServeMux → handler → service]
    Handler -->|Success| Complete[Complete]
    Handler -->|Retryable error| Retry[Retry after a delay]
    Retry --> Redis
    Handler -->|Retries exhausted or skipped| Archive[Archive]

Several workers compete for tasks from the same queue. They do not each receive a copy. Retries and recovery can deliver a task again, so handlers must tolerate repeat execution. For independent reactions to one fact, see events.

Choose the entry point that fits the operation:

The task model, options, handler and administration APIs are native Asynq. The framework adds shared routing, observability, transaction guards and lifecycle ownership. There is no additional Job interface to learn.

Try the existing task#

Use the configured profile from installation. Redis, PostgreSQL, the application migrations and an existing Identity user are required. For automatic signup, enable and configure authentication first. The minimal quickstart keeps authentication off; it cannot create a user through OTP as-is.

Run the combined app, which includes workers by default:

go run ./cmd/app

In another terminal, enqueue the existing welcome task for a real user ID:

go run ./cmd/cli identity welcome '<user-id>'
go run ./cmd/cli notifications list user '<user-id>'

Use the same environment file and database/Redis settings in both terminals. The command prints a task ID after Redis accepts the work. The worker then reloads the account and persists eligible notification channel intents. A worker-owned dispatcher scans those intents every second; delivery workers attempt each channel. The list may initially be empty while that work is pending.

Welcome uses a stable key per recipient and channel. Repeating the command does not create another database welcome notification. Verified addresses also receive email when a mailer is available. The normal app always supplies its configured mailer, but SMTP must be reachable for email to succeed. No cron scheduler is required for notification delivery.

Distinguish the names#

Task types are persisted contracts: renaming one requires a plan for queued tasks. The application uses native Asynq tasks, without a Job interface.

Keep task code with its module#

Pass IDs rather than whole entities, credentials or rendered private data. Reload current account state when the handler runs. Large use cases remain methods on an existing service; creating a task does not require another service abstraction.

Define a task#

Keep the task type, payload and constructor together in internal/modules/<module>/jobs. Use <Operation>Type, <Operation> and New<Operation> consistently. This is the complete existing internal/modules/identity/jobs/welcome.go:

// Package jobs defines Identity's native Asynq tasks and their serializable inputs.
package jobs

import (
	"encoding/json"
	"errors"
	"time"

	"github.com/google/uuid"
	"github.com/hibiken/asynq"

	"github.com/wisalgroup/labena/pkg/asynqx"
)

const WelcomeType = "identity.welcome.v1"

// Welcome carries only identity. The worker reloads account state and contact data;
// names, email addresses, credentials and notification content never enter this job.
type Welcome struct {
	UserID uuid.UUID `json:"user_id"`
}

// NewWelcome constructs a fresh task; enqueue through the app's borrowed client.
// Queue routing comes from application configuration. Retries are safe because
// the notification's occurrence key is stable across attempts.
func NewWelcome(userID uuid.UUID) (*asynq.Task, error) {
	if userID == uuid.Nil {
		return nil, errors.New("identity: welcome job requires a user ID")
	}
	data, err := json.Marshal(Welcome{UserID: userID})
	if err != nil {
		return nil, err
	}
	return asynqx.NewTask(WelcomeType, data, asynq.MaxRetry(5), asynq.Timeout(30*time.Second)), nil
}

asynqx.NewTask initializes headers for client trace propagation. Native options remain available. Create a fresh task for each enqueue; middleware updates its headers, so sharing a task concurrently is unsafe.

Write and register a handler#

A handler decodes the task, calls the owning service or capability, and returns its outcome. Reuse owner contracts such as service.UserService. This complete existing handler lives in internal/modules/identity/delivery/jobs/welcome.go:

// Package jobs handles Identity's native Asynq tasks.
package jobs

import (
	"context"
	"encoding/json"
	"errors"

	"github.com/wisalgroup/labena/internal/modules/shared/entities"

	"github.com/google/uuid"
	"github.com/hibiken/asynq"

	"github.com/wisalgroup/labena/internal/modules/identity/domain"
	identityjobs "github.com/wisalgroup/labena/internal/modules/identity/jobs"
	"github.com/wisalgroup/labena/internal/modules/identity/notifications"
	"github.com/wisalgroup/labena/internal/modules/identity/service"
	"github.com/wisalgroup/labena/internal/modules/shared/enums"
	"github.com/wisalgroup/labena/internal/platform/notification"
)

// Welcome reloads current account state and persists channel intents. Completion
// means notification intent is saved, not that email has reached the recipient.
// Platform notification workers own actual channel delivery and independent retries.
type Welcome struct {
	Users        service.UserService
	Notifier     *notification.Notifier
	EmailEnabled bool
}

func (h Welcome) Handle(ctx context.Context, task *asynq.Task) error {
	var input identityjobs.Welcome
	if task == nil {
		return errors.Join(errors.New("identity: missing welcome job"), asynq.SkipRetry)
	}
	if err := json.Unmarshal(task.Payload(), &input); err != nil || input.UserID == uuid.Nil {
		return errors.Join(errors.New("identity: invalid welcome job"), asynq.SkipRetry)
	}
	user, err := h.Users.Get(ctx, input.UserID)
	if errors.Is(err, domain.ErrUserNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	if user == nil {
		return domain.ErrUserRequired
	}
	if user.DeletedAt != nil || user.Status != enums.UserActive {
		return nil
	}
	content := notifications.NewWelcome(user.Name)
	if !h.EmailEnabled || user.EmailVerifiedAt == nil {
		content.Email = nil
	}
	recipient := notification.Recipient{Type: string(entities.UserEntityType), ID: user.ID.String(), Email: user.Email.String()}
	_, err = h.Notifier.Enqueue(ctx, recipient, content)
	return err
}

Module construction supplies the handler's collaborators. Registration stays in identity/register.go, using the native mux before workers start:

mux.HandleFunc(jobs.WelcomeType, welcome.Handle)

Here jobs imports the task definitions, welcome is the constructed handler, and mux is targets.TaskRouter. A larger business operation belongs on its existing service receiver; transport decoding stays in the handler.

Keep persisted task names and payload fields compatible with work already in Redis. The welcome job saves notification intent; notification workers perform channel delivery. Its stable occurrence key makes repeated attempts converge on the same intents. See notifications for those guarantees.

Enqueue work#

Module assembly passes deps.Queue, the existing asynqx.Client, to operations that need it. No module-specific queue contract is needed. This function-body fragment assumes a borrowed queue, ctx, a UUID userID, and a (string, error) result:

task, err := jobs.NewWelcome(userID)
if err != nil {
    return "", err
}
info, err := queue.EnqueueContext(ctx, task)
if err != nil {
    return "", err
}
return info.ID, nil

Import github.com/wisalgroup/labena/internal/modules/identity/jobs. The returned ID acknowledges queue acceptance; it does not mean the notification has been delivered. Raw enqueue cannot commit atomically with application SQL; see transactions. The app client rejects active framework SQL transaction scopes before enqueue. Standalone clients can install middleware.RequireNoTransaction() outermost; raw native clients bypass this policy. SQL notification intent is available when the notification must commit with a business change.

Enqueue from a service#

HTTP, CLI and event handlers use the same borrowed asynqx.Client method shown above. Pass that dependency to the service that owns the operation during module construction; do not inject the complete Resources bundle into the service. App client middleware already supplies routing, transaction guards and tracing. Constructing a fresh task with asynqx.NewTask prepares its propagation headers.

Keep configuration central#

internal/app/config/queues.go defines the available queues. Both use the same Redis connection and worker pool:

QUEUE_REDIS_URL=redis://127.0.0.1:6379/1
QUEUE_DEFAULT=default
QUEUE_DEFAULT_PRIORITY=1
QUEUE_HIGH_PRIORITY=3
QUEUE_CONCURRENCY=4
QUEUE_ROUTES=identity.welcome.v1=high

A task's destination is resolved in this order:

1. asynq.Queue(...) passed at enqueue time or in schedule options.
2. An exact task-type entry in QUEUE_ROUTES.
3. QUEUE_DEFAULT.

Unlisted task types use the default queue. Unknown queue names fail before publication. Priorities are selection weights, not reserved worker counts. Cache Redis uses database 0 by default; queue Redis uses database 1.

To introduce a third named queue, add it to QueueConfig.loadQueues in internal/app/config/queues.go. Then restart producers and workers with the same definition. QUEUE_ROUTES only selects among configured queues; it does not create a queue definition. Adding a queue name does not create another worker pool.

Configure options and delays#

Native options pass through unchanged, except that app routing owns queue selection:

info, err := queue.EnqueueContext(ctx, task,
    asynq.MaxRetry(3),
    asynq.Timeout(30*time.Second),
    asynq.ProcessIn(5*time.Minute),
    asynq.Retention(24*time.Hour),
)

This fragment imports time and Asynq. Check the error before reading info. ProcessIn delays eligibility once; recurring schedules enqueue new tasks repeatedly. Capacity and polling affect actual start time.

Explicit enqueue options take native precedence over constructor options. Queue options in constructors are overridden by central routing; put queue policy in configuration or enqueue options. Handle asynq.ErrDuplicateTask and asynq.ErrTaskIDConflict with errors.Is when the caller needs to distinguish duplicate acceptance. Uniqueness does not prevent all duplicate effects.

Choose task outcomes#

Reject an invalid task at the handler boundary:

if task == nil || task.Type() != jobs.WelcomeType {
    return errors.Join(errors.New("unexpected task type"), asynq.SkipRetry)
}

This fragment imports errors, Asynq and internal/modules/identity/jobs. Decode and validate its payload before calling the service. Exact-type checks matter because native ServeMux routing also supports prefixes. Do not classify transient infrastructure failures as invalid payloads.

Set retries per task#

In an operation with an injected queue, set native enqueue options:

info, err := queue.EnqueueContext(ctx, task,
    asynq.MaxRetry(3),
    asynq.Timeout(30*time.Second),
)

Handle err before using info. Three retries means up to four ordinary attempts. A timeout cancels cooperative work; it does not reverse an email or database change. MaxRetry(0) disables retries after the initial attempt, not duplicate delivery caused by every possible failure.

Make effects safe to repeat#

Use an operation identity that survives retries. Enforce it with a database unique constraint, conditional transition or downstream idempotency key. Record the effect and its identity atomically where possible. Task uniqueness only reduces duplicate enqueues for a retention window; it does not make an external action atomic with Asynq completion.

A database commit followed by Redis enqueue can fail between the two actions. If both must succeed together, add a deliberate outbox workflow; this app does not provide one automatically.

Identity's welcome handler demonstrates one useful boundary: NewWelcome gives its notification the key welcome, and persisted notification intents enforce recipient/key/channel uniqueness. A repeated task converges on existing intents. The database channel is stored once; an uncertain external email attempt can still duplicate. Choose an idempotency key for the business occurrence, not a fresh random value inside every retry.

Know the boundary#

Asynq is the only queue backend in this app. Redis persistence and recovery settings determine durability. A successful enqueue is not an exactly-once guarantee, and raw SQL changes are not atomic with Redis publication. The notification delivery ledger provides a focused transactional outbox when business operations explicitly use its Enqueue API; it does not make arbitrary tasks transactional.

Keep arbitrary Redis enqueue outside database.TxManager.Execute. The app client returns a transaction error before publishing if the supplied context carries an active framework transaction. A post-commit enqueue can still fail: expose or recover that failure explicitly rather than claiming the SQL write rolled back. For notifications that must be guaranteed by the commit, use the transactional notification workflow.

Read delivery metadata#

Inside a delivered task handler:

id, ok := asynq.GetTaskID(ctx)
if !ok {
    return errors.New("task delivery metadata is missing")
}
queueName, _ := asynq.GetQueueName(ctx)
retries, _ := asynq.GetRetryCount(ctx)
maxRetries, _ := asynq.GetMaxRetry(ctx)

Use these values for diagnostics or an operation that requires them. A direct unit test call has no worker metadata unless explicitly supplied. task.Type(), Payload() and Headers() remain available. Treat delivered task data as read-only. Custom failure classification can preserve the retry counter, so it is not always the total invocation count.

Write a result#

A delivered task has a native ResultWriter. Write a small result before returning:

result, err := json.Marshal(struct {
    Delivered bool `json:"delivered"`
}{Delivered: true})
if err != nil {
    return err
}
_, err = task.ResultWriter().Write(result)
return err

Constructed tasks have no result writer. Do not retain it after the handler returns. Writes replace the result rather than append; Retention keeps successful results available. Store large artifacts elsewhere and return their reference.

Inspect state alongside result bytes: an earlier attempt may have written a result before failing. Result persistence is not atomic with an external effect; a write failure after sending a receipt can cause another send on retry.

Start a worker#

Use the same Redis settings as the producer:

go run ./cmd/worker

This starts Asynq workers, local event subscribers and the notification dispatcher. It opens no business HTTP listener and does not select the cron scheduler. cmd/app already runs workers; use a separate worker process when you want to scale or deploy task execution independently.

Enable the cron scheduler on one designated app process per schedule set. Multiple schedulers would independently enqueue the same recurring work; worker replicas can safely compete for pending deliveries.

Set capacity#

QUEUE_CONCURRENCY=4
QUEUE_DEFAULT_PRIORITY=1
QUEUE_HIGH_PRIORITY=3
QUEUE_STRICT_PRIORITY=false
QUEUE_SHUTDOWN_TIMEOUT=15s

Concurrency is the total number of active tasks per process, across both queues. Three processes configured with 4 can run up to 12 tasks at once. Budget database connections and downstream capacity accordingly.

With weighted selection, high is preferred relative to default; weights do not promise a completion ratio or latency. Strict priority always prefers higher priority work and can starve lower-priority queues. One shared pool is configured; independent pools require explicit native Asynq server wiring in app assembly.

Register before starting#

App construction creates the native ServeMux, installs global worker middleware, and calls module registration. A module's register.go maps job types to handlers:

if targets.TaskRouter != nil && welcome != nil {
    targets.TaskRouter.HandleFunc(jobs.WelcomeType, welcome.Handle)
}

This function-body fragment uses Identity's register.go registration pattern; jobs imports its job definitions and welcome is its constructed delivery handler. It borrows the same user service and notifier as the module. Register every task type that these workers may consume, including app.command when command scheduling is enabled. Do not modify routes after startup.

Readiness and shutdown#

flowchart TB
    accTitle: Keep subscribers alive while workers drain
    accDescr: Worker readiness waits for local event subscribers. On shutdown, stop admission first, drain active tasks next, then drain event handlers before closing shared resources.
    subgraph Startup[Startup dependency]
        Subscribers[Local subscribers ready] --> Worker[Worker verifies Redis and becomes ready]
    end
    subgraph Shutdown[Shutdown order]
        Admission[Stop HTTP, scheduling and dispatch] --> Tasks[Drain active tasks]
        Tasks --> Events[Drain local event handlers]
        Events --> Resources[Close shared resources]
    end

Only selected processes run. In particular, cmd/worker does not start HTTP or the cron scheduler; the shutdown sequence omits services absent from that process.

Redis connectivity is checked before worker readiness. Readiness is a startup signal, not continuous Redis health monitoring. A worker-only process has no /readyz HTTP endpoint; metrics can run on their separate optional listener.

During shutdown, Asynq stops fetching and waits for active handlers within its budget. Handlers must honor cancellation and join their work. Pending Redis tasks remain for another worker; shutdown does not drain the entire backlog. The app closes clients only after their users stop. See lifecycle.

Inspection and administration#

These operator commands are always available:

go run ./cmd/cli queue list
go run ./cmd/cli queue inspect default TASK_ID
go run ./cmd/cli queue pause high
go run ./cmd/cli queue resume high
go run ./cmd/cli queue cancel TASK_ID
go run ./cmd/cli queue archive default TASK_ID
go run ./cmd/cli queue replay default TASK_ID
go run ./cmd/cli queue delete default TASK_ID

Use the task's actual queue. Inspection prints JSON with byte fields encoded as base64. In app assembly, a.inspector.GetTaskInfo(queueName, taskID) returns native TaskInfo with raw Result bytes. The native pagination and bulk APIs remain available. Inspector calls have Redis I/O timeouts, not per-call contexts.

Completed tasks cannot be replayed through native RunTask. Restrict operator access to retained payloads, results and errors. App owns Inspector connections; borrowers must not close them. See safe replay before replaying.

Asynqmon dashboard#

Enable the embedded dashboard independently of CLI administration:

QUEUE_DASHBOARD_ENABLED=true
QUEUE_DASHBOARD_USERNAME=operator
QUEUE_DASHBOARD_PASSWORD=replace-with-a-secret

Restart an HTTP-enabled process and open http://localhost:8080/admin/queues/. The browser requests Basic Auth credentials. UI and API share the existing HTTP listener and use QUEUE_REDIS_URL, including its database index. Use HTTPS for remote access. Disable the setting and restart to remove routes and connections.

Inspect and recover#

Queue administration commands are always available. Inspect before replaying:

go run ./cmd/cli queue list
go run ./cmd/cli queue inspect default TASK_ID
go run ./cmd/cli queue replay default TASK_ID

default must match the task's actual queue. Fix the cause and verify that its business effect can safely repeat. Native replay moves supported scheduled, retry or archived work to pending; it does not reset retry history. Completed tasks need an intentional new enqueue rather than RunTask replay.

The dashboard provides a separate authenticated UI. Pause/resume affects new admission, cancellation is cooperative, and deletion cannot undo effects. See administration for the full command list and state restrictions.

Customize backoff and classification#

After loading config and before app.Serve, optional native callbacks can change retry timing and whether an error counts toward failure statistics:

cfg.Queue.RetryDelay = func(n int, err error, task *asynq.Task) time.Duration {
    limit := time.Second << min(n, 6)
    return limit/2 + time.Duration(rand.Int64N(int64(limit/2)))
}

This uses time, math/rand/v2 and Asynq. It bounds exponential delay with jitter. Keep callbacks fast and safe for concurrent calls. Nil callbacks keep native defaults.

cfg.Queue.IsFailure = func(err error) bool {
    return !errors.Is(err, ErrTemporarilyThrottled)
}

Define ErrTemporarilyThrottled in your application. Returning false does not make the task successful: while retries remain it can retry without incrementing the counter. Repeated non-failures can therefore continue indefinitely. Define a terminal policy; classification does not override an already exhausted budget.

Tracing and metrics#

app/infrastructure.go installs global client routing and tracing, native worker middleware, and Asynq's official Prometheus collector. Workers create consumer spans linked to producer context. Registered mux patterns bound task-name metric labels; IDs belong in logs and traces. Baggage, payloads and results are not copied into routine telemetry.

Native recurring schedules bypass the client chain. Routing is resolved at registration; each execution still has a worker span without a stale startup trace. See observability.

Inspect capacity and failures#

Enable the queue dashboard to inspect pending, active, delayed, retry and archived tasks. With metrics enabled, the official Asynq collector exposes queue state and pending latency; handler duration records each attempt. Long queues can mean insufficient capacity, slow dependencies or paused workers—not necessarily failed publication.

Use failure handling to define retries and replay safely.

Troubleshoot pending or repeated tasks#

Task payloads, results and retained errors are operator data. Restrict Redis, dashboard and backup access, especially for queued mail. Avoid logging bodies, credentials or arbitrary provider error text in application telemetry.

Test tasks and their integration#

Unit-test decoding, invalid payloads, missing records and repeat-safe effects by calling handlers directly with a task and test collaborators. Such calls have no worker metadata or result writer. Keep business tests under tests/<module>; delivery adapters and reusable queue middleware may have tests beside their owners.

go test ./pkg/asynqx/... ./internal/modules/identity/delivery/jobs/...
go test ./tests/modules -run TestWorkflowGuidesMatchCompiledSources -count=1

Those checks do not prove Redis delivery. The database and Redis integration gate exercises signup → event → task → notification storage. Use task testing for cancellation, retries and lifecycle tests.

Use queues independently#

pkg/asynqx can be used without app assembly. A native *asynq.Client satisfies asynqx.Client; the caller owns its connection and worker setup. This function-body fragment assumes ctx, userID and options (asynq.RedisClientOpt) are supplied:

client := asynq.NewClient(options)
defer client.Close()

task, err := jobs.NewWelcome(userID)
if err != nil {
    return err
}
_, err = client.EnqueueContext(ctx, task, asynq.Queue("high"))
return err

The raw native client does not install the framework's routing, tracing or SQL transaction guard. To retain the guard with native enqueueing, wrap once using github.com/wisalgroup/labena/pkg/asynqx/middleware:

queue := middleware.Chain(client.EnqueueContext, middleware.RequireNoTransaction())

Route(routing) and ObserveClient(provider) are optional members of that same chain. They are application assembly concerns; business callers still use the same EnqueueContext method.

asynqx.NewWorker(server, handler, redisPool, startupTimeout) borrows its clients and checks Redis when Run(ctx) starts. Register handlers first, run once, wait for shutdown to finish, then close owned clients. Handlers must cooperate with cancellation: Go cannot forcibly terminate a goroutine. Do not also call native Start, Run or Shutdown on a server managed by this adapter.

Continue with recurring schedules, events, or the complete onboarding workflow.