README
¶
queue
A Valkey/asynq foundation for
background-job systems: connection config, a task client verified with a PING
at construction, a common enqueue-option subset that covers most use cases,
and an event Envelope wire format (with OTel trace propagation).
queue has no dependency beyond
github.com/gp-system/errs, so it works
in any Go program built on asynq and go-redis (Valkey is RESP/command
compatible with Redis, so the go-redis client works against it unchanged).
It is also the queue foundation used by the
gp-system backend kit's events, scheduler,
outbox and worker layers.
The problem it solves
A background-job system built on asynq needs a few decisions made once and
shared everywhere: what the Valkey connection config looks like, what a task
type string means (is it an event, a listener, a scheduled job?), how a task
payload carries a trace context across the enqueue/process hop, and what
"duplicate" means when idempotent delivery is required. queue centralizes
those decisions once, so the different layers of a background-job system
(event dispatch, scheduling, outbox relay, worker processing) and
application code build on the same primitives instead of reinventing them
per project. asynq types (*asynq.Task, *asynq.TaskInfo) are not hidden:
application code that needs the full asynq API is free to use it directly
alongside this package.
Install
go get github.com/gp-system/queue
Usage
Config and Client
Config describes the shared Valkey connection; compose it under a prefix:
type Config struct {
Valkey queue.Config `envPrefix:"VALKEY_"`
}
which maps to VALKEY_ADDR, VALKEY_PASSWORD, VALKEY_DB. NewClient opens
the connection, pings it, and builds an asynq client:
client, err := queue.NewClient(ctx, cfg.Valkey)
if err != nil {
log.Fatal(err)
}
defer client.Close()
info, err := client.Enqueue(ctx, task, queue.OnQueue("emails"), queue.MaxRetry(3))
if errors.Is(err, queue.ErrDuplicate) {
// a TaskID/Unique constraint suppressed the enqueue; treat as success
}
MustNewClient is NewClient but panics on error, for use in main().
Options
Option re-exports the useful subset of asynq's enqueue options so common
cases don't require importing asynq directly: OnQueue, MaxRetry,
Timeout, Deadline, ProcessIn, Unique, Retention, TaskID.
AsynqOptions converts them to raw []asynq.Option for code that calls
asynq directly (e.g. a scheduler registrar) rather than through
Client.Enqueue.
Task type naming
Every task type is one of four prefixed shapes, so a single asynq.ServeMux
can classify any task type it sees without a side channel:
| Prefix | Constructor | Meaning |
|---|---|---|
event: |
EventTaskType(name) |
an event's fan-out task |
listener: |
ListenerTaskType(event, listener) |
one listener's copy of an event |
job: |
JobTaskType(name) |
a scheduled job |
schedule: |
ScheduleTaskType(name) |
a scheduled event trigger |
IsEventTaskType, IsListenerTaskType, IsJobTaskType, IsScheduleTaskType
classify a task type string; JobName extracts the job name back out of a
job task type.
Envelope
Envelope is the wire format of every event task payload: it travels from
the dispatcher (or outbox) through the fan-out task to each listener task,
carrying the same id, payload and trace context throughout.
env := queue.Envelope{
ID: id,
Name: "news.published",
Payload: payload,
OccurredAt: time.Now().UTC(),
}
env.InjectTrace(ctx)
task, err := env.Task()
DecodeEnvelope extracts the envelope back out of a task on the consuming
side; ExtractTrace returns a context carrying the trace context stored in
the envelope's metadata, so spans link across the dispatch, relay and worker
hop.
Design rules
- Centralize the wire format, don't hide asynq. The task-type prefixes and envelope shape are the single source of truth every layer (and application code) reads and writes against; asynq itself is never wrapped away.
- Duplicates are a sentinel, not the raw asynq error.
Enqueuemapsasynq.ErrTaskIDConflict/asynq.ErrDuplicateTasktoErrDuplicate. Wrapping a sentinel error is idiomatic Go, not a package-specific convention: callers check it witherrors.Isinstead of matching the underlying asynq error type directly. - The client pings at construction. A broken Valkey connection fails at
startup, not on the first enqueue, matching
pg.NewPool's convention.
Documentation
¶
Overview ¶
Package queue is the kit's Valkey/asynq foundation: connection config, a task client verified with a PING at construction, the enqueue-option subset the rest of the kit uses, and the event Envelope wire format (with OTel trace propagation). The events, scheduler, outbox and worker packages build on it. The point is centralizing this core once — task-type prefixes, envelope shape, trace propagation — so it isn't reinvented per project; asynq types (e.g. *asynq.Task, *asynq.TaskInfo) are not hidden and application code may use this package and asynq directly when events/scheduler don't fit.
Index ¶
- Constants
- Variables
- func AsynqOptions(opts ...Option) []asynq.Option
- func EventTaskType(name string) string
- func IsEventTaskType(t string) bool
- func IsJobTaskType(t string) bool
- func IsListenerTaskType(t string) bool
- func IsScheduleTaskType(t string) bool
- func JobName(t string) (string, bool)
- func JobTaskType(name string) string
- func ListenerTaskType(event, listener string) string
- func ScheduleTaskType(name string) string
- type Client
- type Config
- type Envelope
- type Option
Constants ¶
const ( EventTaskPrefix = "event:" ListenerTaskPrefix = "listener:" JobTaskPrefix = "job:" ScheduleTaskPrefix = "schedule:" )
Task type prefixes. Event dispatch enqueues an EventTaskType; the worker's fan-out handler expands it into one ListenerTaskType per registered listener. Scheduled jobs use JobTaskType. Exported so a asynq.ServeMux is registered against the same constants this package uses to classify task types — there must be exactly one source of truth for the wire prefixes.
Variables ¶
var ErrDuplicate = errors.New("queue: duplicate task")
ErrDuplicate is returned by Enqueue when a TaskID or Unique constraint suppressed the enqueue because an identical task already exists. Callers that rely on idempotent delivery (the outbox relay, event fan-out) treat it as success.
Functions ¶
func AsynqOptions ¶
AsynqOptions converts kit options to raw asynq options, for code that calls asynq directly (the scheduler registrar) rather than through Client.Enqueue.
func EventTaskType ¶
EventTaskType is the asynq task type for an event's fan-out task.
func IsEventTaskType ¶
IsEventTaskType reports whether a task type is an event fan-out task.
func IsJobTaskType ¶
IsJobTaskType reports whether a task type is a scheduled job.
func IsListenerTaskType ¶
IsListenerTaskType reports whether a task type is a listener task.
func IsScheduleTaskType ¶
IsScheduleTaskType reports whether a task type is a scheduled event trigger.
func JobName ¶
JobName returns the job name encoded in a job task type, and false if t is not a job task type.
func JobTaskType ¶
JobTaskType is the asynq task type for a scheduled job.
func ListenerTaskType ¶
ListenerTaskType is the asynq task type for one listener of an event.
func ScheduleTaskType ¶
ScheduleTaskType is the asynq task type for a scheduled event trigger. The worker rebuilds a fresh envelope per fire (keyed on the trigger's task id) and fans it out, so each scheduled fire delivers to every listener once.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client wraps an asynq.Client over one shared go-redis connection to Valkey. The connection is verified with a PING at construction, matching the kit convention (pg.NewPool pings the pool).
func MustNewClient ¶
MustNewClient is NewClient but panics on error. Intended for main().
func (*Client) Close ¶
Close releases the underlying Valkey connection. The asynq client is built from this shared connection and does not own it (asynq refuses to close a shared client), so closing the connection is the complete teardown.
func (*Client) Enqueue ¶
func (c *Client) Enqueue(ctx context.Context, task *asynq.Task, opts ...Option) (*asynq.TaskInfo, error)
Enqueue submits a task to Valkey. It returns ErrDuplicate (per the kit convention of wrapping a sentinel, not the underlying asynq error) when a TaskID/Unique constraint suppressed the enqueue.
type Config ¶
type Config struct {
// Addr is the Valkey host:port.
Addr string `env:"ADDR" envDefault:"localhost:6379"`
// Password authenticates the connection; empty for no auth.
Password string `env:"PASSWORD"`
// DB selects the Valkey logical database.
DB int `env:"DB" envDefault:"0"`
}
Config describes the Valkey connection shared by the queue client, worker, scheduler and outbox relay. Compose it under a prefix:
type Config struct {
Valkey queue.Config `envPrefix:"VALKEY_"`
}
which maps to VALKEY_ADDR, VALKEY_PASSWORD, VALKEY_DB.
func (Config) ValkeyConnOpt ¶
func (c Config) ValkeyConnOpt() asynq.RedisClientOpt
ValkeyConnOpt renders the config as asynq's connection option, used to build the client, server and scheduler.
func (Config) ValkeyOptions ¶
ValkeyOptions renders the config as go-redis connection options (Valkey is RESP/command compatible with Redis), the single source of truth for every direct redis.NewClient call in the kit (queue, scheduler, ...).
type Envelope ¶
type Envelope struct {
// ID is a unique identifier for this dispatch (also used as the asynq
// TaskID for deduplication). Listeners use it as an idempotency key.
ID string `json:"id"`
// Name is the event name (Event.EventName()).
Name string `json:"name"`
// Payload is the JSON-encoded event value.
Payload json.RawMessage `json:"payload"`
// Metadata carries the W3C trace context (traceparent/tracestate) so spans
// link across the dispatch → relay → worker hop.
Metadata map[string]string `json:"metadata,omitempty"`
// OccurredAt is when the event was dispatched.
OccurredAt time.Time `json:"occurred_at"`
}
Envelope is the wire format of every event task payload. It travels from the dispatcher (or outbox) through the fan-out task to each listener task, so the listener sees the same id, payload and trace context the producer set.
func DecodeEnvelope ¶
DecodeEnvelope extracts the envelope from a task payload.
func (*Envelope) ExtractTrace ¶
ExtractTrace returns a context carrying the trace context stored in the envelope metadata (a no-op when none was injected).
func (*Envelope) InjectTrace ¶
InjectTrace writes the trace context carried by ctx into the envelope metadata, so downstream consumers can continue the trace.
type Option ¶
type Option func(*taskOptions)
Option customizes how a task is enqueued. The kit re-exports the useful subset of asynq's enqueue options so common cases don't require importing asynq directly; application code that needs the full option set is free to use AsynqOptions/FromAsynqOptions and asynq types directly — the goal is a centralized, shared building block, not hiding asynq.
func OnQueue ¶
OnQueue routes the task to a named queue (matched against the worker's WORKER_QUEUES weights). Unset means the "default" queue.
Source Files