Queue

What it is for

gas/queue sends and receives background jobs through AWS SQS, or through an SQS-compatible service such as ElasticMQ or LocalStack. Its sqs.Service implements gas.JobQueueProvider, the interface declared in core, so the code that enqueues and consumes jobs imports only github.com/gasmod/gas and the backend is chosen where the app is wired.

The interface is pull-based. Gas does not run a consumer for you: your code calls Dequeue in its own loop, decides the concurrency and the failure policy, and settles each job with Ack or Nack. A queue is named by its full queue URL, and the package never creates queues.

Package	Provides
queue	The sentinel ErrClosed
queue/sqs	Service, the SQS backend
queue/queuetest	MockQueue, a recording mock of gas.JobQueueProvider

Install

go get github.com/gasmod/gas/queue

Wiring

cmd/worker/main.go

w := gas.NewWorker(
  gas.WithServiceInstance[gas.ConfigProvider](cfg),
  gas.WithSingletonService[gas.Logger](gaslog.NewSlogLogger()),

  // The SQS backend, under the interface services depend on.
  gas.WithSingletonService[gas.JobQueueProvider](queuesqs.New()),

  // An app service that consumes jobs; see Consuming below.
  gas.WithSingletonService[*Consumer](NewConsumer),
)

sqs.New() returns a constructor that resolves gas.ConfigProvider and gas.Logger from the container. It takes two options:

• sqs.WithConfig(cfg) uses cfg as the whole configuration and skips the ConfigProvider. Start from sqs.DefaultConfig() so the keys you do not set keep their defaults. A config given this way is used as given, without validation.
• sqs.WithClient(client) injects a ready SQS client; Init then builds none and fetches no credentials.

A queue works the same in a gas.App and in a gas.Worker. The example uses a Worker because a consumer usually runs without an HTTP server; see Workers for the Worker itself, including running it on AWS Lambda.

To point the backend at a local ElasticMQ without touching the environment, pass a config:

// elasticMQ points the backend at a local ElasticMQ with static credentials.
// WithConfig replaces the whole Config and skips the ConfigProvider, so it
// starts from DefaultConfig to keep the other defaults.
func elasticMQ() gas.Option {
  cfg := queuesqs.DefaultConfig()
  cfg.Queue.Endpoint = "http://localhost:9324"
  cfg.Queue.AccessKeyID = "x"
  cfg.Queue.SecretAccessKey = "x"

  return gas.WithSingletonService[gas.JobQueueProvider](
    queuesqs.New(queuesqs.WithConfig(cfg)),
  )
}

Usage

Enqueueing

// SendWelcome enqueues one job. The queue is named by its full URL.
func SendWelcome(ctx context.Context, q gas.JobQueueProvider, queueURL string, userID string) error {
  return q.Enqueue(ctx, queueURL, []byte(`{"user_id":"`+userID+`"}`),
    gas.WithDelay(10*time.Second),
    gas.WithJobAttributes(map[string]string{"kind": "welcome"}),
  )
}

Enqueue sends the payload as the message body. The options are declared in core and map onto the SQS SendMessage request:

Option	SQS field	Notes
gas.WithDelay(d)	DelaySeconds	Whole seconds, truncated. Set only when d is positive
gas.WithGroupID(id)	MessageGroupId	Set only when id is not empty
gas.WithDedupeID(id)	MessageDeduplicationId	Set only when id is not empty
gas.WithJobAttributes(m)	MessageAttributes	Each entry becomes a String attribute

Gas passes the group and deduplication IDs on any queue and leaves the checks to SQS. They are meant for FIFO queues:

// On a FIFO queue (its URL ends in .fifo), the group orders messages and
// the deduplication ID drops repeats.
func SendReceipt(ctx context.Context, q gas.JobQueueProvider, fifoURL, orderID string) error {
  return q.Enqueue(ctx, fifoURL, []byte(orderID),
    gas.WithGroupID("order-"+orderID),
    gas.WithDedupeID("receipt-"+orderID),
  )
}

Consuming

A consumer is an ordinary service that owns its loop. This one starts the loop in Init and stops it in Close:

jobs/consumer.go

// ConsumerConfig is bound from the ConfigProvider (key jobs.queueurl).
type ConsumerConfig struct {
  Jobs struct {
    QueueURL string `validate:"required"`
  }
}

// Consumer runs a pull loop for the life of the service.
type Consumer struct {
  queue  gas.JobQueueProvider
  logger gas.Logger
  url    string

  cancel context.CancelFunc
  done   sync.WaitGroup
}

func NewConsumer(q gas.JobQueueProvider, cfg gas.ConfigProvider, logger gas.Logger) (*Consumer, error) {
  var c ConsumerConfig
  if err := cfg.Bind(&c); err != nil {
    return nil, err
  }
  return &Consumer{queue: q, logger: logger, url: c.Jobs.QueueURL}, nil
}

func (c *Consumer) Name() string { return "jobs-consumer" }

// Init starts the loop on its own context: the ctx Init receives is
// canceled as soon as Init returns.
func (c *Consumer) Init(context.Context) error {
  ctx, cancel := context.WithCancel(context.Background())
  c.cancel = cancel
  c.done.Go(func() { c.run(ctx) })
  return nil
}

// Close stops the loop and waits for the batch in hand. It runs before the
// queue's own Close, since the consumer depends on the queue.
func (c *Consumer) Close(context.Context) error {
  c.cancel()
  c.done.Wait()
  return nil
}

func (c *Consumer) run(ctx context.Context) {
  for ctx.Err() == nil {
    if err := c.Poll(ctx); err != nil {
      if ctx.Err() != nil || errors.Is(err, queue.ErrClosed) {
        return
      }
      c.logger.Error("poll failed").Err("error", err).Send()
      select {
      case <-ctx.Done():
      case <-time.After(5 * time.Second):
      }
    }
  }
}

Two details in it matter. The ctx that Init receives is canceled as soon as Init returns, so the loop runs on a context of its own. And because the consumer takes gas.JobQueueProvider as a dependency, it is initialized after the queue and closed before it, so Close drains the batch in hand while the queue still accepts Ack and Nack.

Each pass receives a batch and settles every job in it:

// Poll receives up to 10 jobs, waiting up to 20 seconds for the first, and
// settles each one: Ack removes it, Nack makes it visible again at once.
func (c *Consumer) Poll(ctx context.Context) error {
  jobs, err := c.queue.Dequeue(ctx, c.url, 10, 20*time.Second)
  if err != nil {
    return err
  }

  for _, job := range jobs {
    if err := handle(ctx, job); err != nil {
      c.logger.Warn("job failed, retrying").Str("job_id", job.ID).Err("error", err).Send()
      if err := c.queue.Nack(ctx, c.url, job); err != nil {
        return err
      }
      continue
    }
    if err := c.queue.Ack(ctx, c.url, job); err != nil {
      return err
    }
  }
  return nil
}

Dequeue(ctx, queueURL, maxMessages, wait) long-polls SQS ReceiveMessage:

• maxMessages is clamped to SQS’s range of 1 to 10. A value above 10 also logs a warning.
• wait is sent in whole seconds. When it truncates to zero, the configured queue.waittimeseconds is used instead. A nonzero wait is passed as given, so keep it at 20 seconds or less, the most SQS allows.
• The configured queue.visibilitytimeout is applied to every message received.

Each gas.Job carries the SQS message ID in ID, the receipt handle in ReceiptHandle, and the body in Body. Attributes merges the message’s system attributes with its message attributes that have a string value; on a shared name the message attribute wins. Attributes is nil when the message has neither.

Ack deletes the message. Nack sets its visibility timeout to zero, so it can be received again at once. A job that is neither acked nor nacked becomes visible again when its visibility timeout ends.

Make handlers idempotent

Delivery, ordering, and retries are SQS’s. A process that dies after doing the work but before Ack leaves the message to reappear when its visibility timeout ends, and the job runs again. Key side effects on something stable in the payload.

Using the SQS client directly

Service.Client() returns the underlying *sqs.Client for calls the interface does not cover, such as creating or purging a queue:

// The backend's SQS client covers what JobQueueProvider does not, such as
// creating a queue.
func createQueue(ctx context.Context, svc *queuesqs.Service, name string) (string, error) {
  out, err := svc.Client().CreateQueue(ctx, &awssqs.CreateQueueInput{QueueName: &name})
  if err != nil {
    return "", err
  }
  return *out.QueueUrl, nil
}

Client returns nil until Init has built the client, and when WithClient injected a client that is not an *sqs.Client.

Configuration

Unless WithConfig is given, Init binds the settings from the gas.ConfigProvider over sqs.DefaultConfig(), so a key that is not set keeps its default. The keys sit under queue:

Key	Type	Default	Description
queue.region	string	us-east-1	AWS region. Must not be empty
queue.endpoint	string	empty	Custom endpoint URL, for ElasticMQ or LocalStack. Empty uses the AWS endpoint
queue.accesskeyid	string	empty	Static access key ID. Used only when queue.secretaccesskey is set too; otherwise the default AWS credential chain applies
queue.secretaccesskey	string	empty	Static secret access key
queue.visibilitytimeout	duration	30s	How long a received message stays hidden from other consumers. Must not be negative
queue.waittimeseconds	int	20	Long-poll duration used when Dequeue is given no wait. 0 to 20

The rules in the table are validate tags, which ConfigProvider.Bind enforces when it binds the config; a violation fails Init. How keys map to environment variables is described on Config.

Init loads the AWS configuration and then fetches the credentials, both under Init’s context, so a missing or broken credential setup fails startup rather than the first call. With the default chain this can make a network call, for example to the instance metadata service. Init does not contact SQS: an unreachable endpoint, rejected credentials, or a missing queue surface on the first operation.

Errors

Error	Returned by	When
queue.ErrClosed	Enqueue, Dequeue, Ack, Nack, CheckReady	After Close
gas/queue/sqs: config binding: ...	Init	The config cannot be bound or fails its validate tags
gas/queue/sqs: load aws config: ...	Init	The AWS configuration cannot be loaded
gas/queue/sqs: retrieve aws credentials: ...	Init	No credentials can be found
gas/queue/sqs: enqueue: ... (and dequeue, ack, nack)	The matching operation	The SQS call failed

queue.ErrClosed is the only sentinel. Errors from SQS calls wrap the AWS SDK’s error with %w, so errors.As reaches the SDK’s error types.

Known issue KI-036

Enqueue, Dequeue, Ack, and Nack on a service that was constructed but never initialized dereference a nil client and panic instead of returning an error. In an app the container runs Init before handing the service out, so this affects only code that calls the constructor by hand. See Known issues.

Health and readiness

sqs.Service implements gas.ReadyReporter, not gas.HealthReporter. CheckReady returns gas/queue/sqs: not initialized before Init has built a client and queue.ErrClosed after Close, so a readiness probe takes the instance out of rotation while it drains. It does not call SQS.

Close only marks the service closed: the service starts no goroutines and has nothing else to release. A later Init, as RestartService runs, clears the mark.

Testing

queuetest.MockQueue implements gas.JobQueueProvider. Each method records the call in Calls (a Method name and its Args) and then delegates to the matching EnqueueFn, DequeueFn, AckFn, or NackFn field. With the field unset it returns zero values: a nil error, and no jobs from Dequeue. CallCount(method) counts the calls to one method, and Reset clears them.

jobs/consumer_test.go

import (
  "context"
  "testing"
  "time"

  "github.com/gasmod/gas"
  "github.com/gasmod/gas/queue/queuetest"
)

// TestPollSettlesEachJob drives one Poll against the mock: the good job is
// acked, the empty one nacked, and SQS is never involved.
func TestPollSettlesEachJob(t *testing.T) {
  t.Parallel()
  q := &queuetest.MockQueue{
    DequeueFn: func(context.Context, string, int, time.Duration) ([]gas.Job, error) {
      return []gas.Job{
        {ID: "1", Body: []byte(`{"user_id":"42"}`)},
        {ID: "2"},
      }, nil
    },
  }
  c := &Consumer{queue: q, logger: &gas.NopLogger{}, url: "test-queue"}

  if err := c.Poll(context.Background()); err != nil {
    t.Fatalf("Poll: %v", err)
  }
  if got := q.CallCount("Ack"); got != 1 {
    t.Errorf("Ack calls = %d, want 1", got)
  }
  if got := q.CallCount("Nack"); got != 1 {
    t.Errorf("Nack calls = %d, want 1", got)
  }
}

To test against a real queue without AWS, run ElasticMQ or LocalStack and set queue.endpoint to it.

API reference

• github.com/gasmod/gas/queue
• github.com/gasmod/gas/queue/sqs
• github.com/gasmod/gas/queue/queuetest
• gas.JobQueueProvider and its options, in core