README
¶
Arkoren Queue SQS
arkoren.dev/driver/aws/awsqueue provides an Amazon SQS queue driver. It publishes
and consumes complete queue.Message envelopes. Consumers accept both direct
SQS envelopes and standard SNS notification wrappers, so one publisher can
fan out to multiple SQS worker queues through SNS.
Install with go get arkoren.dev/driver/aws and import
"arkoren.dev/driver/aws/awsqueue". Provision the SQS queues first. The
SQS queue guide
covers AWS permissions, facade wiring, visibility limits, and local emulation.
The configured logical queue is authoritative on delivery. SQS supplies the message ID and receive count; untrusted envelope values do not select a worker queue or override broker metadata. Retry rounds positive delays up to whole seconds and rejects negative delays or delays above 12 hours without settlement. SQS also enforces the time remaining since receipt, so even a 12-hour request can fail. The driver returns that native error; it never shortens the requested delay to claim success. Zero requests immediate visibility.
Fail optionally sends a failure envelope before
deleting the original message; this favors redelivery when failure publication
does not succeed. Ordinary dispatch and failure queues support FIFO destinations:
messages group by logical queue and deduplicate by message ID. Malformed failure
reports without an envelope use the failed group and a fresh deduplication ID.
Failure archives contain job payloads and error text; secure them like the source
queue rather than treating them as ordinary diagnostic logs.
Pass optional settings as trailing func(*awsqueue.Builder) callbacks:
driver, err := awsqueue.New(ctx, map[string]string{
"emails": "https://sqs.eu-west-1.amazonaws.com/123456789012/emails",
}, func(consumer *awsqueue.Builder) {
consumer.MaxMessages(5)
})
Callbacks run directly once, synchronously in order on one fresh draft. Nil
entries panic; expanded nil/empty slices omit edits. Settings return that draft;
publication copies the completed settings and required queue map. Retaining a
draft or changing the original map cannot change a driver. All settings also
apply to NewFrom(client, queues, configure...); the client is not copied or closed.
VisibilityTimeout(0) selects 30 seconds. Other values must be from one second
through 12 hours inclusive; native requests round fractions upward, while the
heartbeat runs at half the configured duration. Invalid settings fail before
AWS configuration loading. Nil Logger selects slog.Default().
Each renewal has a complete SDK budget of at most ten seconds and at most half the configured lease. Renewal continues while a canceled consumer drains a handler. When the handler returns, its pending renewal is canceled and joined. Settlement waits behind pending renewal using its own context budget. This is cooperative cancellation: native transports, credentials, middleware, and handlers must honor their contexts. Renewal failure or SQS's remaining-visibility limit can still permit duplicate delivery; handlers must be idempotent.
Documentation
¶
Overview ¶
Package awsqueue provides an Amazon SQS-backed queue driver.
New loads the default AWS configuration. NewFrom accepts a caller-owned SDK client. Both accept ordered Builder callbacks invoked directly once on one fresh draft. Nil entries panic; expanded nil/empty slices omit edits. Publication copies the required routing map and completed settings, not clients/loggers. Queue mappings and completed settings are validated before New loads AWS configuration; NewFrom performs no network I/O. VisibilityTimeout defaults to 30 seconds when zero, otherwise it must be one second through 12 hours inclusive. Requests round fractional seconds up; renewal ticks use half the configured duration. SQS can still reject renewal beyond a message's remaining visibility lifetime; configuration validation does not guarantee renewal. Retry rejects delays above 12 hours rather than silently shortening them; native remaining-time failures leave settlement available for another attempt. FIFO dispatch and failure reporting retain the existing JSON envelopes and supply native group/deduplication fields. Failure archives contain payloads and handler error text and require the same access protection as source jobs.
Renewal operations have a budget of half the visibility or ten seconds, whichever is shorter, including SDK retries. They continue during canceled handler drain. Handler return cancels in-flight renewal before joining its worker; no worker is abandoned. Application settlement waits for prior renewal without exceeding its own context budget. Native clients and their dependencies must honor operation contexts for cancellation to release I/O. Renewal failures are logged, not treated as successful settlement; duplicate delivery remains possible. As with other RPC errors, cancellation does not prove SQS ignored the request. See Driver.Consume for context and settlement wiring.
Example:
driver, err := awsqueue.New(ctx, map[string]string{"emails": "queue-url"}, func(consumer *awsqueue.Builder) {
consumer.MaxMessages(5)
})
if err != nil {
return err
}
_ = driver
Index ¶
- Variables
- type Builder
- func (consumer *Builder) FailureQueueURL(url string) *Builder
- func (consumer *Builder) Logger(logger *slog.Logger) *Builder
- func (consumer *Builder) MaxMessages(count int32) *Builder
- func (consumer *Builder) VisibilityTimeout(timeout time.Duration) *Builder
- func (consumer *Builder) WaitTimeSeconds(seconds int32) *Builder
- type BuilderFunc
- type Driver
Constants ¶
This section is empty.
Variables ¶
var ( // ErrUnknownQueue is returned when a queue name has no configured SQS URL. // No message is sent or received for that unknown mapping. // // Example: // // if errors.Is(err, awsqueue.ErrUnknownQueue) { // return fmt.Errorf("configure the logical queue destination: %w", err) // } ErrUnknownQueue = errors.New("unknown job queue") // ErrInvalidConfig is returned when SQS driver configuration is invalid. // Required routing and polling/visibility bounds are checked at construction. // // Example: // // if errors.Is(err, awsqueue.ErrInvalidConfig) { // return fmt.Errorf("correct queue construction settings: %w", err) // } ErrInvalidConfig = errors.New("invalid SQS job driver configuration") // ErrUnsupportedRetryDelay is returned when an SQS visibility timeout cannot // represent the requested retry delay. // A rejected delay never settles the delivery or shortens the requested wait. // // Example: // // if errors.Is(err, awsqueue.ErrUnsupportedRetryDelay) { // return fmt.Errorf("choose a delay supported by SQS: %w", err) // } ErrUnsupportedRetryDelay = errors.New("unsupported SQS job retry delay") // ErrInvalidDelivery is returned when SQS does not provide the data required // to settle a received delivery. // Missing receipt handles cannot be acknowledged by the driver. // // Example: // // if errors.Is(err, awsqueue.ErrInvalidDelivery) { // return fmt.Errorf("invalid native SQS delivery: %w", err) // } ErrInvalidDelivery = errors.New("invalid SQS job delivery") // ErrInvalidHandler is returned when Consume is given a nil handler. // A handler must explicitly settle each accepted delivery. // // Example: // // if errors.Is(err, awsqueue.ErrInvalidHandler) { // return fmt.Errorf("supply a delivery handler before consuming: %w", err) // } ErrInvalidHandler = errors.New("invalid job delivery handler") // ErrDeliverySettled is returned when a delivery is settled more than once. // Concurrent settlement admission also fails with this sentinel. // // Example: // // if errors.Is(err, awsqueue.ErrDeliverySettled) { // return fmt.Errorf("delivery already has a settlement owner: %w", err) // } ErrDeliverySettled = errors.New("job delivery already settled") )
Functions ¶
This section is empty.
Types ¶
type Builder ¶
type Builder struct {
// contains filtered or unexported fields
}
Builder holds optional SQS construction settings. Callbacks share one fresh draft initialized with polling defaults; publication does not retain it.
Example:
configure := func(consumer *Builder) {
consumer.MaxMessages(5)
}
driver, err := New(ctx, map[string]string{"emails": "queue-url"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) FailureQueueURL ¶
FailureQueueURL replaces the terminal-report destination. Empty disables reports. The same draft is returned; no queue is created.
Example:
configure := func(consumer *Builder) {
consumer.FailureQueueURL("")
}
driver, err := New(ctx, map[string]string{"emails": "queue-url"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) Logger ¶
Logger replaces the shared delivery logger. Nil selects the driver's existing default logging policy. It returns the draft without changing a live logger.
Example:
configure := func(consumer *Builder) {
consumer.Logger(slog.New(slog.DiscardHandler))
}
driver, err := New(ctx, map[string]string{"emails": "queue-url"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) MaxMessages ¶
MaxMessages replaces the pull batch limit. Zero selects one; construction rejects values outside 1–10. It returns the same construction draft.
Example:
configure := func(consumer *Builder) {
consumer.MaxMessages(5)
}
driver, err := New(ctx, map[string]string{"emails": "queue-url"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) VisibilityTimeout ¶
VisibilityTimeout replaces the requested lease duration. Zero selects the 30-second default, not immediate visibility or disabled renewal. New and NewFrom reject nonzero values outside one second through 12 hours inclusive with ErrInvalidConfig before loading AWS configuration or making requests. Fractional seconds round up for SQS; heartbeat ticks use half the configured duration, not half the rounded value. Each renewal SDK operation is bounded by that half duration or ten seconds, whichever is shorter. Renewal errors are logged and can allow redelivery; see Driver.Consume. It returns the same draft.
Example:
configure := func(consumer *Builder) {
consumer.VisibilityTimeout(time.Minute)
}
driver, err := New(ctx, map[string]string{"emails": "queue-url"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) WaitTimeSeconds ¶
WaitTimeSeconds replaces the long-poll duration in seconds. Zero selects 20; construction rejects values outside 1–20. It returns the same draft.
Example:
configure := func(consumer *Builder) {
consumer.WaitTimeSeconds(1)
}
driver, err := New(ctx, map[string]string{"emails": "queue-url"}, configure)
if err != nil {
return err
}
_ = driver
type BuilderFunc ¶
type BuilderFunc = func(*Builder)
BuilderFunc edits one fresh SQS construction draft in argument order. Nil entries panic; publication detaches completed settings, not native clients.
Example:
var configure awsqueue.BuilderFunc = func(consumer *awsqueue.Builder) {
consumer.VisibilityTimeout(time.Minute)
}
driver, err := awsqueue.New(ctx, map[string]string{"emails": "queue-url"}, configure)
if err != nil {
return err
}
_ = driver
type Driver ¶
type Driver struct {
// contains filtered or unexported fields
}
Driver implements queue.DispatcherDriver and queue.ConsumerDriver using Amazon SQS. Native clients are borrowed; each Consume call owns its per-delivery heartbeats.
Example:
driver, err := awsqueue.NewFrom(client, map[string]string{"emails": "queue-url"})
if err != nil {
return err
}
dispatcher := queue.NewDispatcher(driver)
_ = dispatcher
func New ¶
New creates the driver with credentials and region from the default AWS configuration chain (environment, shared config, instance role).
queues maps application queue names to SQS queue URLs. It is required: an empty map, or any entry with a blank name or URL, returns an error wrapping ErrInvalidConfig. The map is copied, so later changes by the caller do not affect the driver. configure callbacks adjust optional settings directly once, synchronously in order on one fresh draft. Nil entries panic; expanded nil or empty slices supply no callbacks. Retained drafts cannot change the driver. Queue mappings and completed settings are validated before AWS configuration loading. VisibilityTimeout accepts zero (30 seconds) or one second through 12 hours inclusive; other values wrap ErrInvalidConfig. Fractional seconds round up in SQS requests; the heartbeat uses half the configured duration.
Example:
driver, err := awsqueue.New(ctx, map[string]string{
"emails": "https://sqs.eu-west-1.amazonaws.com/123456789012/emails",
}, func(consumer *awsqueue.Builder) {
consumer.MaxMessages(5)
})
if err != nil {
return err
}
_ = driver
func NewFrom ¶
func NewFrom(client *sqs.Client, queues map[string]string, configure ...BuilderFunc) (*Driver, error)
NewFrom creates an SQS job driver using client. The caller retains ownership of client.
queues is required and validated and copied exactly as in New; a nil client also returns an error wrapping ErrInvalidConfig. All Builder settings apply, including New's visibility bounds and zero default. Validation makes no SQS requests and starts no goroutines. Callbacks follow New's ordered once-only policy, including nil panics. Native HTTP clients, credential providers, and middleware must honor operation contexts: the driver cancels and joins heartbeat I/O rather than abandoning goroutines when a supplied dependency ignores cancellation.
Example:
driver, err := awsqueue.NewFrom(sqs.NewFromConfig(cfg), map[string]string{
"emails": "https://sqs.eu-west-1.amazonaws.com/123456789012/emails",
})
if err != nil {
return err
}
_ = driver
func (*Driver) Client ¶
Client returns the underlying SQS client, or nil when the driver was built from a test double. Native policy remains shared; direct queue operations bypass this driver's envelope, lease, and settlement handling.
Example:
if native := driver.Client(); native != nil {
_ = native.Options().Region
}
func (*Driver) Consume ¶
func (driver *Driver) Consume(ctx context.Context, queueName string, handler queue.DeliveryHandler) error
Consume receives messages from queue until ctx is canceled or SQS returns an error. Handler errors are logged and left unsettled for SQS redelivery. It waits for handlers to return; cancellation does not abandon them. Renewal continues during handler drain, with each complete SDK call limited to the lesser of ten seconds and half the configured visibility. Failed renewals are logged and can permit redelivery; handlers must tolerate duplicate delivery.
Once a handler returns, its heartbeat I/O is canceled before its worker is joined. Application settlement serializes behind any pending renewal, but waiting honors the settlement context. Native dependencies must honor request contexts for canceled I/O and joining to finish; see NewFrom.
Example:
driver, err := awsqueue.NewFrom(sqs.NewFromConfig(cfg), map[string]string{
"emails": "https://sqs.eu-west-1.amazonaws.com/123456789012/emails",
})
if err != nil {
return err
}
err = driver.Consume(ctx, "emails", func(ctx context.Context, received queue.Delivery) error {
settleContext, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
defer cancel()
return received.Acknowledge(settleContext)
})
if err != nil && !errors.Is(err, context.Canceled) {
return err
}
func (*Driver) Dispatch ¶
Dispatch serializes message and sends it to the configured URL for its queue. FIFO queues group by logical queue and deduplicate on the supplied ID, or a fresh ID when omitted. Success reports acceptance, not handler completion.
Example:
message := queue.Message{
ID: "job-1", Name: "send-email", Queue: "emails", Payload: []byte(`{"id":1}`),
}
if err := driver.Dispatch(ctx, message); err != nil {
return err
}
func (*Driver) MessagingSystem ¶
MessagingSystem returns "aws_sqs", the messaging.system telemetry value that the queue and broadcast facades report for this driver.
Example:
var driver awsqueue.Driver name := driver.MessagingSystem() _ = name
Source Files
Directories