From f08db1f71cec9b6af513b016f054d0dc7f1c3461 Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Tue, 29 Sep 2026 11:35:02 +0100 Subject: [PATCH 01/11] feat(PAR-2054): duplicate v1 into v2 so we can make breaking changes --- v2/config.go | 118 ++++++++++++++ v2/consumers.go | 261 +++++++++++++++++++++++++++++++ v2/controller.go | 24 +++ v2/deleter.go | 63 ++++++++ v2/go.mod | 18 +++ v2/go.sum | 20 +++ v2/internal/client/client.go | 20 +++ v2/internal/messages/messages.go | 191 ++++++++++++++++++++++ v2/message.go | 9 ++ v2/retriever.go | 58 +++++++ v2/sqs.go | 139 ++++++++++++++++ v2/worker.go | 110 +++++++++++++ v2/worker_test.go | 251 +++++++++++++++++++++++++++++ 13 files changed, 1282 insertions(+) create mode 100644 v2/config.go create mode 100644 v2/consumers.go create mode 100644 v2/controller.go create mode 100644 v2/deleter.go create mode 100644 v2/go.mod create mode 100644 v2/go.sum create mode 100644 v2/internal/client/client.go create mode 100644 v2/internal/messages/messages.go create mode 100644 v2/message.go create mode 100644 v2/retriever.go create mode 100644 v2/sqs.go create mode 100644 v2/worker.go create mode 100644 v2/worker_test.go diff --git a/v2/config.go b/v2/config.go new file mode 100644 index 0000000..18e254d --- /dev/null +++ b/v2/config.go @@ -0,0 +1,118 @@ +package formigo + +import ( + "log" + "time" + + "github.com/Pod-Point/go-queue-worker/v2/internal/client" +) + +const ( + defaultErrorThreshold = 3 + defaultErrorPeriod = time.Second * 120 + defaultConcurrency = 100 + defaultRetrievers = 1 + defaultDeleterBufferSize = 10 + defaultDeleterBufferTimeout = time.Millisecond * 500 +) + +type DeleterConfiguration struct { + BufferSize int + BufferTimeout time.Duration +} + +// The ErrorConfiguration defines a threshold for which the worker stops. If the number +// of errors occurred during the worker execution passes the given Threshold on the +// specified Period, the worker stops. +type ErrorConfiguration struct { + // Number of errors that must occur in the Period before the worker stops. + // Default: 3. + Threshold int + + // Duration of the period for which, if the number of errors passes the Threshold, the worker stops. + // Default: 120s. + Period time.Duration + + // The error report function, returns a boolean value to decide whether the error counts towards to threshold + ReportFunc func(err error) +} + +// The BatchConsumerBufferConfiguration defines a buffer which is consumed by the worker when either +// the buffer is full or the timeout has passed since the first message got added. +type BatchConsumerBufferConfiguration struct { + // Max number of messages that the buffer can contain. + // Default: 10. + Size int + + // Time after which the buffer gets processed, no matter whether it is full or not. + // This value MUST be smaller tha VisibilityTimeout in the + // RetrieveMessageConfiguration + the maximum processing time of the handler. + // If this is not set correctly, the same message could be processed multiple times. + // Default: 1s. + Timeout time.Duration +} + +type MessageConsumerConfiguration struct { + Handler MessageHandler +} + +type BatchConsumerConfiguration struct { + Handler BatchHandler + BufferConfig BatchConsumerBufferConfiguration +} + +type Configuration struct { + // A queue client + Client client.Client + + // Number of Go routines that process the messages from the Queue. + // The higher this value, the more Go routines are spawned to process the messages. + // Using a high value can be useful when the Handler of the consumer perform slow I/O operations. + // Default: 100. + Concurrency int + + // Number of Go routines that retrieve messages from the Queue. + // The higher this value, the more Go routines are spawned to read the messages from the + // queue and provide them to the worker's consumers. + // Using a high value can be useful when the network is slow or when consumers are quicker + // than retrievers. + // Default: 1. + Retrievers int + + // The ErrorConfiguration. + ErrorConfig ErrorConfiguration + + // The messages Consumer. + Consumer Consumer + + // Configuration for the deleter + DeleterConfig DeleterConfiguration +} + +func setWorkerConfigValues(config Configuration) Configuration { + if config.Retrievers == 0 { + config.Retrievers = defaultRetrievers + } + + if config.Concurrency == 0 { + config.Concurrency = defaultConcurrency + } + + if config.ErrorConfig.ReportFunc == nil { + config.ErrorConfig.ReportFunc = func(err error) bool { + log.Println("ERROR", err) + + return true + } + } + + if config.DeleterConfig.BufferSize == 0 { + config.DeleterConfig.BufferSize = defaultDeleterBufferSize + } + + if config.DeleterConfig.BufferTimeout == 0 { + config.DeleterConfig.BufferTimeout = defaultDeleterBufferTimeout + } + + return config +} diff --git a/v2/consumers.go b/v2/consumers.go new file mode 100644 index 0000000..e45ddb5 --- /dev/null +++ b/v2/consumers.go @@ -0,0 +1,261 @@ +package formigo + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/Pod-Point/go-queue-worker/v2/internal/messages" +) + +type BatchResponse struct { + FailedMessagesId []interface{} +} + +type MessageHandler = func(ctx context.Context, msg Message) error +type BatchHandler = func(ctx context.Context, msgs []Message) (BatchResponse, error) + +// This means that the buffered messages didn't get passed to the handler within +// the first message's timeout. +// This is generally due to: +// - Visibility timeout of the messages too small +// - Buffer timeout too high +// - Consumer to slow +var errBufferCtxExpired = errors.New("buffer context expired, buffer will Reset") + +type Consumer interface { + consume(concurrency int, ctrl *controller, messageCh <-chan messages.Message, deleteCh chan<- messages.Message) +} + +func makeAvailableConsumers(concurrency int) chan struct{} { + consumers := make(chan struct{}, concurrency) + for i := 0; i < concurrency; i++ { + consumers <- struct{}{} + } + + return consumers +} + +// wrapHandler catches any panic error and returns the error that generated it. +// It prevents the worker from crashing in case of an unexpected error. +func wrapHandler(handler func() error) (err error) { + defer func() { + if r := recover(); r != nil { + // Set error before returning + err = fmt.Errorf("panic error: %s", r) + } + }() + + err = handler() + + return err +} + +// messageConsumer defines a message handler that consumes only one message at a +// time. +// It can be useful when the workload is specific per message, for example for sending +// an email. +type messageConsumer struct { + handler MessageHandler +} + +func (c *messageConsumer) processMessage(msg messages.Message) error { + defer msg.CancelCtx() // This must be called to release resources associated with the context. + + // Process Message + return wrapHandler(func() error { + return c.handler(msg.Ctx, msg) + }) +} + +// Consumes and deletes a single message, it stops only when the `messageCh` gets closed +// and doesn't have any messages in it. +func (c *messageConsumer) consume(concurrency int, ctrl *controller, messageCh <-chan messages.Message, deleteCh chan<- messages.Message) { + consumers := makeAvailableConsumers(concurrency) + + var wg sync.WaitGroup + for msg := range messageCh { + <-consumers // Use an available consumer + + wg.Add(1) + go func(message messages.Message) { + defer func() { + wg.Done() + consumers <- struct{}{} // Release consumer + }() + + err := c.processMessage(message) + if err != nil { + ctrl.reportError(fmt.Errorf("failed to process message: %w", err)) + return + } + + // Push message for deletion + deleteCh <- message + }(msg) + } + + wg.Wait() +} + +func NewMessageConsumer(config MessageConsumerConfiguration) *messageConsumer { + return &messageConsumer{ + handler: config.Handler, + } +} + +// batchConsumer allows to process multiple messages at a time. This can be useful +// for batch updates or use cases with high throughput. +type batchConsumer struct { + handler BatchHandler + bufferConfig BatchConsumerBufferConfiguration +} + +// It processes the messages and push them downstream for deletion. +func (c *batchConsumer) processMessages(ctrl *controller, deleteCh chan<- messages.Message, ctx context.Context, msgs []messages.Message) { + defer func() { + if r := recover(); r != nil { + ctrl.reportError(fmt.Errorf("panic error: %s", r)) + } + }() + + // Convert slice to the abstraction + converted := make([]Message, 0, len(msgs)) + for _, msg := range msgs { + converted = append(converted, msg) + } + + resp, err := c.handler(ctx, converted) + if err != nil { + ctrl.reportError(fmt.Errorf("failed to process batch: %w", err)) + } + + toDelete := c.buildMessagesToDeleteFromBatchResponse(msgs, resp) + // Push messages for deletion + for _, msg := range toDelete { + deleteCh <- msg + } +} + +// Consumes and deletes a number of messages in the interval [1, N] based on configuration +// provided in the BufferConfiguration. +// It stops only when the messageCh gets closed and doesn't have any messages in it. +func (c *batchConsumer) consume(concurrency int, ctrl *controller, messageCh <-chan messages.Message, deleteCh chan<- messages.Message) { + consumers := makeAvailableConsumers(concurrency) + + // Create buffer + buffer := messages.NewBufferWithContextTimeout(messages.BufferWithContextTimeoutConfiguration{ + Size: c.bufferConfig.Size, + BufferTimeout: c.bufferConfig.Timeout, + }) + defer buffer.Reset() + + var wg sync.WaitGroup + func() { + for { + select { + case msg, open := <-messageCh: + if !open { + // Message channel closed. This is the stop signal. + // Note: We can't return if the buffer contains messages to process. + // We MUST consume all the messages on the buffer + if buffer.IsEmpty() { + return // Buffer empty, we can stop + } + break // Buffer contains messages, break the select + } + + // Add message to the buffer + buffer.Add(msg) + + // If the buffer is not full, continue + if !buffer.IsFull() { + continue + } + + case <-buffer.CtxExpired(): + ctrl.reportError(errBufferCtxExpired) + + // Reset the buffer. + buffer.Reset() + continue + + case <-buffer.Expired(): + // Timeout expired, process the buffer + } + + select { + case <-consumers: // Use an available consumer + case <-buffer.CtxExpired(): + ctrl.reportError(errBufferCtxExpired) + + // Reset the buffer. + buffer.Reset() + continue + } + + ctx, cancelCtx := buffer.PullContext() + + wg.Add(1) + go func(ctx context.Context, ctxCancelFunc context.CancelFunc, msgs []messages.Message) { + defer func() { + wg.Done() + consumers <- struct{}{} // Release consumer + ctxCancelFunc() // Cancel context + }() + + // Process the messages + c.processMessages(ctrl, deleteCh, ctx, msgs) + }(ctx, cancelCtx, buffer.Messages()) + + // Reset buffer + buffer.Reset() + } + }() + + wg.Wait() +} + +func (c *batchConsumer) buildMessagesToDeleteFromBatchResponse(msgs []messages.Message, resp BatchResponse) []messages.Message { + if len(resp.FailedMessagesId) == 0 { + return msgs + } + + toDelete := make([]messages.Message, 0, len(msgs)) + + failedMessagesIdIndexed := make(map[interface{}]struct{}, len(resp.FailedMessagesId)) + for _, id := range resp.FailedMessagesId { + failedMessagesIdIndexed[id] = struct{}{} + } + + for _, msg := range msgs { + if _, ok := failedMessagesIdIndexed[msg.Id()]; !ok { + toDelete = append(toDelete, msg) + } + } + + return toDelete +} + +func NewBatchConsumer(config BatchConsumerConfiguration) *batchConsumer { + if config.BufferConfig.Size == 0 { + config.BufferConfig.Size = 10 + } + + if config.BufferConfig.Timeout == 0 { + config.BufferConfig.Timeout = time.Second + } + + return &batchConsumer{ + handler: config.Handler, + bufferConfig: config.BufferConfig, + } +} + +// Interface guards +var ( + _ Consumer = (*messageConsumer)(nil) + _ Consumer = (*batchConsumer)(nil) +) diff --git a/v2/controller.go b/v2/controller.go new file mode 100644 index 0000000..429120b --- /dev/null +++ b/v2/controller.go @@ -0,0 +1,24 @@ +package formigo + +import ( + "context" + "sync" +) + +type controller struct { + errorConfig ErrorConfiguration + errorCounter int + mutex sync.Mutex + stopOnce sync.Once + stopFunc context.CancelCauseFunc +} + +func (c *controller) reportError(err error) { + c.errorConfig.ReportFunc(err) +} + +func newController(errorConfig ErrorConfiguration) *controller { + return &controller{ + errorConfig: errorConfig, + } +} diff --git a/v2/deleter.go b/v2/deleter.go new file mode 100644 index 0000000..f392ff0 --- /dev/null +++ b/v2/deleter.go @@ -0,0 +1,63 @@ +package formigo + +import ( + "fmt" + "sync" + + "github.com/Pod-Point/go-queue-worker/v2/internal/client" + "github.com/Pod-Point/go-queue-worker/v2/internal/messages" +) + +// deleter will delete messages from SQS until the delete channel gets closed. +// Any error will be sent to the error channel. +func deleter(wg *sync.WaitGroup, deleter client.MessageDeleter, config DeleterConfiguration, ctrl *controller, deleteCh <-chan messages.Message) { + // Create buffer + buffer := messages.NewMessageBuffer(messages.BufferConfiguration{ + Size: config.BufferSize, + Timeout: config.BufferTimeout, + }) + defer buffer.Reset() + + for { + select { + case msg, open := <-deleteCh: + if !open { + // Delete channel closed. This is the stop signal. + // Note: We can't return if the buffer contains messages to delete. + // We MUST send the deletion request if the buffer has any messages + if buffer.IsEmpty() { + return // Buffer empty, we can stop + } + break // Buffer contains messages, break the select + } + + buffer.Add(msg) + + // If the buffer is not full, continue + if !buffer.IsFull() { + continue + } + + case <-buffer.Expired(): + // Buffer expired, process the buffer + } + + // Get the messages + msgs := buffer.Messages() + + // Send deletion request in a separate Go routine to maximise the throughput + wg.Add(1) + go func(msgs []messages.Message) { + defer wg.Done() + // No context should be passed here. In fact, if the messages were processed correctly + // we want to do our best to delete it from the queue. + err := deleter.DeleteMessages(msgs) + if err != nil { + ctrl.reportError(fmt.Errorf("unable to delete %d messages: %w", len(msgs), err)) + } + }(msgs) + + // Reset buffer + buffer.Reset() + } +} diff --git a/v2/go.mod b/v2/go.mod new file mode 100644 index 0000000..cee5512 --- /dev/null +++ b/v2/go.mod @@ -0,0 +1,18 @@ +module github.com/Pod-Point/go-queue-worker/v2 + +go 1.26 + +require ( + github.com/aws/aws-sdk-go-v2/service/sqs v1.34.5 + github.com/stretchr/testify v1.9.0 +) + +require ( + github.com/aws/aws-sdk-go-v2 v1.30.4 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.16 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.16 // indirect + github.com/aws/smithy-go v1.20.4 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/v2/go.sum b/v2/go.sum new file mode 100644 index 0000000..5c94707 --- /dev/null +++ b/v2/go.sum @@ -0,0 +1,20 @@ +github.com/aws/aws-sdk-go-v2 v1.30.4 h1:frhcagrVNrzmT95RJImMHgabt99vkXGslubDaDagTk8= +github.com/aws/aws-sdk-go-v2 v1.30.4/go.mod h1:CT+ZPWXbYrci8chcARI3OmI/qgd+f6WtuLOoaIA8PR0= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.16 h1:TNyt/+X43KJ9IJJMjKfa3bNTiZbUP7DeCxfbTROESwY= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.16/go.mod h1:2DwJF39FlNAUiX5pAc0UNeiz16lK2t7IaFcm0LFHEgc= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.16 h1:jYfy8UPmd+6kJW5YhY0L1/KftReOGxI/4NtVSTh9O/I= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.16/go.mod h1:7ZfEPZxkW42Afq4uQB8H2E2e6ebh6mXTueEpYzjCzcs= +github.com/aws/aws-sdk-go-v2/service/sqs v1.34.5 h1:HYyVDOC2/PIg+3oBX1q0wtDU5kONki6lrgIG0afrBkY= +github.com/aws/aws-sdk-go-v2/service/sqs v1.34.5/go.mod h1:7idt3XszF6sE9WPS1GqZRiDJOxw4oPtlRBXodWnCGjU= +github.com/aws/smithy-go v1.20.4 h1:2HK1zBdPgRbjFOHlfeQZfpC4r72MOb9bZkiFwggKO+4= +github.com/aws/smithy-go v1.20.4/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/v2/internal/client/client.go b/v2/internal/client/client.go new file mode 100644 index 0000000..0af07a8 --- /dev/null +++ b/v2/internal/client/client.go @@ -0,0 +1,20 @@ +package client + +import ( + "context" + + "github.com/Pod-Point/go-queue-worker/v2/internal/messages" +) + +type MessageReceiver interface { + ReceiveMessages(ctx context.Context) ([]messages.Message, error) +} + +type MessageDeleter interface { + DeleteMessages(messages []messages.Message) error +} + +type Client interface { + MessageReceiver + MessageDeleter +} diff --git a/v2/internal/messages/messages.go b/v2/internal/messages/messages.go new file mode 100644 index 0000000..ba91502 --- /dev/null +++ b/v2/internal/messages/messages.go @@ -0,0 +1,191 @@ +package messages + +import ( + "context" + "time" +) + +type Message struct { + Ctx context.Context `json:"-"` // Exclude from JSON + CancelCtx context.CancelFunc `json:"-"` // Exclude from JSON + MsgId interface{} `json:"id"` + Msg interface{} `json:"content"` + ReceivedTime time.Time `json:"receivedAt"` +} + +func (m Message) Id() interface{} { + return m.MsgId +} + +func (m Message) Content() interface{} { + return m.Msg +} + +func (m Message) ReceivedAt() time.Time { + return m.ReceivedTime +} + +type BufferConfiguration struct { + Size int + Timeout time.Duration +} + +// Buffer is used to implement a buffer with a size and timeout. +// When the buffer is full, the `full()` method returns true. +// If the buffer has expired, the `timer` will emit. +// Note that buffer is not thread safe. Remember to use a mutex +// in case it's been used by multiple Go routines. +type Buffer struct { + messages []Message + + // expiredCh emits when the buffer expires. + // When the buffer is initialised or reset the channel is set + // so that it never expires. + expiredCh <-chan time.Time + + size int + timeout time.Duration + timer *time.Timer +} + +// Len returns the number of Messages in the buffer. +func (b *Buffer) Len() int { + return len(b.messages) +} + +// IsFull Returns true if the buffer is full. +func (b *Buffer) IsFull() bool { + return len(b.messages) == b.size +} + +// Add a message to the buffer and set the timer if the buffer was empty. +func (b *Buffer) Add(msg Message) { + if len(b.messages) == 0 { + // Create a new timer and assign its channel to expiredCh + b.timer = time.NewTimer(b.timeout) + b.expiredCh = b.timer.C + } + + b.messages = append(b.messages, msg) +} + +// IsEmpty returns true if the buffer is empty. +func (b *Buffer) IsEmpty() bool { + return len(b.messages) == 0 +} + +// Messages returns the messages held in the buffer. +func (b *Buffer) Messages() []Message { + return b.messages +} + +// Reset flush the messages contained in the buffer and stop the timer. +// It's important to stop the timer to avoid memory leaks. In fact, the +// GC won't collect the timer until its channel expires. +// NOTE: this function should be always called to clean up any buffer +// created. Used in defer can guarantee that it always run. +func (b *Buffer) Reset() { + if b.timer != nil { + // Stop the timer to free its resources + b.timer.Stop() + b.timer = nil + + // Reset the expiredCh so that it never expires + b.expiredCh = make(<-chan time.Time) + } + + if len(b.messages) > 0 { + b.messages = make([]Message, 0, b.size) + } +} + +// Expired emits when the buffer expires. +func (b *Buffer) Expired() <-chan time.Time { + return b.expiredCh +} + +func NewMessageBuffer(config BufferConfiguration) *Buffer { + return &Buffer{ + expiredCh: make(<-chan time.Time), + messages: make([]Message, 0, config.Size), + size: config.Size, + timeout: config.Timeout, + timer: nil, + } +} + +type BufferWithContextTimeoutConfiguration struct { + BufferTimeout time.Duration + CtxTimeout time.Duration + Size int +} + +// BufferWithContextTimeout is used to construct a buffer that has a context timeout +// along with the standard buffer timeout. This is used because the messages have to +// be processed within a certain period and if this doesn't happen, the buffer should +// delete the messages in it and reset. +type BufferWithContextTimeout struct { + *Buffer + ctx context.Context + cancelCtx context.CancelFunc +} + +func (b *BufferWithContextTimeout) Add(msg Message) { + if len(b.messages) == 0 { + // Set the context of the buffer to first message's context + b.ctx = msg.Ctx + } + + // Override the current cancelCtx in a way that cancels all + // the previous messages' contexts. + b.cancelCtx = func(cancel context.CancelFunc) context.CancelFunc { + return func() { + cancel() + msg.CancelCtx() + } + }(b.cancelCtx) + + b.Buffer.Add(msg) +} + +// Reset resets its internal buffer, cancel the current context created and +// reset any timeout. +// It's important to call this function to avoid memory leaks. In fact, the +// GC won't collect any timer or resources allocated within the context. +// NOTE: this function should be always called to clean up any buffer +// created. Used in defer can guarantee that it always run. +func (b *BufferWithContextTimeout) Reset() { + b.Buffer.Reset() + + b.cancelCtx() // Be sure to reset any previous context + b.ctx = context.Background() // Create a context that doesn't expire + b.cancelCtx = func() {} +} + +func (b *BufferWithContextTimeout) CtxExpired() <-chan struct{} { + return b.ctx.Done() +} + +func (b *BufferWithContextTimeout) Context() context.Context { + return b.ctx +} + +func (b *BufferWithContextTimeout) PullContext() (context.Context, context.CancelFunc) { + ctx, cancelCtx := b.ctx, b.cancelCtx + + b.ctx = context.Background() // Create a context that doesn't expire + b.cancelCtx = func() {} + + return ctx, cancelCtx +} + +func NewBufferWithContextTimeout(config BufferWithContextTimeoutConfiguration) *BufferWithContextTimeout { + return &BufferWithContextTimeout{ + Buffer: NewMessageBuffer(BufferConfiguration{ + Size: config.Size, + Timeout: config.BufferTimeout, + }), + ctx: context.Background(), + cancelCtx: func() {}, // Empty cancelCtx + } +} diff --git a/v2/message.go b/v2/message.go new file mode 100644 index 0000000..d1b118a --- /dev/null +++ b/v2/message.go @@ -0,0 +1,9 @@ +package formigo + +import "time" + +type Message interface { + ReceivedAt() time.Time + Content() interface{} + Id() interface{} +} diff --git a/v2/retriever.go b/v2/retriever.go new file mode 100644 index 0000000..e3da536 --- /dev/null +++ b/v2/retriever.go @@ -0,0 +1,58 @@ +package formigo + +import ( + "context" + "errors" + "fmt" + + "github.com/Pod-Point/go-queue-worker/v2/internal/client" + "github.com/Pod-Point/go-queue-worker/v2/internal/messages" +) + +// retriever will get messages from SQS until the given context gets canceled. +// Any error will be sent to the controller. +func retriever(ctx context.Context, receiver client.MessageReceiver, ctrl *controller, messageCh chan<- messages.Message) { + for { + select { + case <-ctx.Done(): + return + default: + msgs, err := receiver.ReceiveMessages(ctx) + if err != nil { + if errors.Is(err, context.Canceled) && errors.Is(ctx.Err(), context.Canceled) { + // The worker's context was canceled. We can exit. + return + } + + // Report the error to the controller and continue. + ctrl.reportError(fmt.Errorf("unable to receive message: %w", err)) + continue + } + + // All the messages retrieved must be processed. + // This means that the retriever won't listen for context cancellation + // at this stage. + func() { + for _, msg := range msgs { + select { + case <-msg.Ctx.Done(): + // If consumers don't pick up the messages within the messages' timeout we raise + // an error. + // This could be due to one or more of the following reasons: + // - message timeout too small. + // - consumer too slow. Increasing the number of consumers may help, especially if + // the handler performs many I/O operations. + // + // Note that we won't process all messages retrieved by the API calls. This is because + // the visibility timeout is the same for all the messages returned by the call. + ctrl.reportError(errors.New("message didn't get picked up by any consumer within its timeout")) + + return // Avoid publishing all the messages downstream + case messageCh <- msg: + // Message pushed to the channel + } + } + }() + } + } +} diff --git a/v2/sqs.go b/v2/sqs.go new file mode 100644 index 0000000..268fcf4 --- /dev/null +++ b/v2/sqs.go @@ -0,0 +1,139 @@ +package formigo + +import ( + "context" + "fmt" + "strconv" + "time" + + awsSqs "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/aws/aws-sdk-go-v2/service/sqs/types" + + "github.com/Pod-Point/go-queue-worker/v2/internal/client" + "github.com/Pod-Point/go-queue-worker/v2/internal/messages" +) + +type SqsClientConfiguration struct { + // The AWS Sqs Service Client + Svc *awsSqs.Client + + // The AWS ReceiveMessageInput + ReceiveMessageInput *awsSqs.ReceiveMessageInput + + // Defines the interval within which the message must be processed. + // If empty, it tries to set the value from the ReceiveMessageInput's + // VisibilityTimeout. + // If ReceiveMessageInput's VisibilityTimeout is empty it retrieves the + // default value set on the queue. This action will fail if the client + // does not have the permission to retrieve the SQS queue's attributes. + MessageCtxTimeout time.Duration +} + +type sqsClient struct { + svc *awsSqs.Client + receiveMessageInput *awsSqs.ReceiveMessageInput + messageCtxTimeout time.Duration +} + +func (c sqsClient) ReceiveMessages(ctx context.Context) ([]messages.Message, error) { + out, err := c.svc.ReceiveMessage(ctx, c.receiveMessageInput) + if err != nil { + return nil, fmt.Errorf("unable to receive messages: %w", err) + } + + msgs := make([]messages.Message, len(out.Messages)) + for i, msg := range out.Messages { + msgs[i] = c.createMessage(msg) + } + + return msgs, nil +} + +func (c sqsClient) DeleteMessages(messages []messages.Message) error { + _, err := c.svc.DeleteMessageBatch(context.Background(), &awsSqs.DeleteMessageBatchInput{ + Entries: c.prepareMessagesForDeletion(messages), + QueueUrl: c.receiveMessageInput.QueueUrl, + }) + + return err +} + +// prepareMessagesForDeletion takes the processed message batch to transform into a de-duplicated struct for SQS to handle +func (c sqsClient) prepareMessagesForDeletion(messages []messages.Message) []types.DeleteMessageBatchRequestEntry { + deleteEntries := make([]types.DeleteMessageBatchRequestEntry, 0, len(messages)) + processed := map[string]bool{} + + for _, message := range messages { + msgId := message.Msg.(types.Message).MessageId + if _, exists := processed[*msgId]; exists { + continue + } + + deleteEntries = append(deleteEntries, types.DeleteMessageBatchRequestEntry{ + Id: msgId, + ReceiptHandle: message.Msg.(types.Message).ReceiptHandle, + }) + + processed[*msgId] = true + } + + return deleteEntries +} + +func (c sqsClient) createMessage(sqsMessage types.Message) messages.Message { + msg := messages.Message{ + MsgId: *sqsMessage.MessageId, + Msg: sqsMessage, + ReceivedTime: time.Now(), + } + + // Set a context with timeout + msg.Ctx, msg.CancelCtx = context.WithTimeout(context.Background(), c.messageCtxTimeout) + + return msg +} + +func NewSqsClient(ctx context.Context, config SqsClientConfiguration) (sqsClient, error) { + messageTimeout := config.MessageCtxTimeout + + // Try config.ReceiveMessageInput.VisibilityTimeout first + if messageTimeout == 0 && config.ReceiveMessageInput.VisibilityTimeout != 0 { + messageTimeout = time.Second * time.Duration(config.ReceiveMessageInput.VisibilityTimeout) + } + // Otherwise, infer it from SQS queue's VisibilityTimeout attribute + if messageTimeout == 0 { + var err error + messageTimeout, err = retrieveSqsQueueVisibilityTimeout(ctx, config.Svc, config.ReceiveMessageInput.QueueUrl) + if err != nil { + return sqsClient{}, fmt.Errorf("unable to retrieve visibility timeout: %w", err) + } + } + + return sqsClient{ + svc: config.Svc, + receiveMessageInput: config.ReceiveMessageInput, + messageCtxTimeout: messageTimeout, + }, nil +} + +func retrieveSqsQueueVisibilityTimeout(ctx context.Context, svc *awsSqs.Client, queue *string) (time.Duration, error) { + out, err := svc.GetQueueAttributes(ctx, &awsSqs.GetQueueAttributesInput{ + QueueUrl: queue, + AttributeNames: []types.QueueAttributeName{types.QueueAttributeNameVisibilityTimeout}, + }) + if err != nil { + return 0, fmt.Errorf("unable to get attributes: %w", err) + } + + timeout, err := strconv.Atoi(out.Attributes[string(types.QueueAttributeNameVisibilityTimeout)]) + if err != nil { + return 0, fmt.Errorf("unable to parse timeout value: %w", err) + } + + return time.Second * time.Duration(timeout), nil +} + +// Interface guards +var ( + _ client.Client = (*sqsClient)(nil) +) diff --git a/v2/worker.go b/v2/worker.go new file mode 100644 index 0000000..654a732 --- /dev/null +++ b/v2/worker.go @@ -0,0 +1,110 @@ +package formigo + +import ( + "context" + "errors" + "sync" + + "github.com/Pod-Point/go-queue-worker/v2/internal/client" + "github.com/Pod-Point/go-queue-worker/v2/internal/messages" +) + +type Worker struct { + client client.Client + concurrency int + retrievers int + errorConfig ErrorConfiguration + consumer Consumer + deleterConfig DeleterConfiguration +} + +func (w Worker) Run(ctx context.Context) error { + // Create a new context with a cancel function used to stop the worker from the + // controller in case too many errors occur. + ctx, cancel := context.WithCancelCause(ctx) + + // Create controller + ctrl := newController(w.errorConfig, cancel) + + // Run retrievers and get the message channel + messageCh := w.runRetrievers(ctx, ctrl) + + // Run consumer and get the deletion channel. + // Note that the context is not given to them because they will only stop once + // all the messages in the pipeline have been consumed. + deleteCh := w.runConsumer(ctrl, messageCh) + + // Run deleter. + // Note that the context is not given to the deleter because it will only stop once + // all the consumed messages in the pipeline have been deleted. + var wg sync.WaitGroup // WaitGroup for the deleter + deleter(&wg, w.client, w.deleterConfig, ctrl, deleteCh) + + // Wait for deleter to exit + wg.Wait() + + // Get the cause of the cancellation + // If the context was cancelled by the controller, there must be a cause containing + // the error, otherwise the context error by default is `context.Canceled`, which + // means that it was a normal stop request (probably SIGTERM). + err := context.Cause(ctx) + if errors.Is(err, context.Canceled) { + return nil + } + + return err +} + +// runRetrievers will run a number of retrievers (Go routines) based on the worker's +// configuration. +// It returns a channel where the messages will be published and, only when all the +// retrievers have stopped, it will close it to broadcast the signal to stop to the +// consumers. +func (w Worker) runRetrievers(ctx context.Context, ctrl *controller) <-chan messages.Message { + messageCh := make(chan messages.Message) + + var wg sync.WaitGroup + wg.Add(w.retrievers) + for i := 0; i < w.retrievers; i++ { + go func() { + defer wg.Done() + retriever(ctx, w.client, ctrl, messageCh) + }() + } + + go func() { + wg.Wait() + close(messageCh) + }() + + return messageCh +} + +// runConsumer runs the worker's consumer. +// It returns a channel where the messages will be published for deletion and, +// only when the consumer has stopped, it will close it to broadcast the +// signal to stop to the deleter. +func (w Worker) runConsumer(ctrl *controller, messageCh <-chan messages.Message) <-chan messages.Message { + deleteCh := make(chan messages.Message) + + go func() { + w.consumer.consume(w.concurrency, ctrl, messageCh, deleteCh) + + close(deleteCh) + }() + + return deleteCh +} + +func NewWorker(config Configuration) Worker { + config = setWorkerConfigValues(config) + + return Worker{ + client: config.Client, + concurrency: config.Concurrency, + retrievers: config.Retrievers, + errorConfig: config.ErrorConfig, + consumer: config.Consumer, + deleterConfig: config.DeleterConfig, + } +} diff --git a/v2/worker_test.go b/v2/worker_test.go new file mode 100644 index 0000000..e00bada --- /dev/null +++ b/v2/worker_test.go @@ -0,0 +1,251 @@ +package formigo + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + + "github.com/Pod-Point/go-queue-worker/v2/internal/messages" +) + +type SimpleInMemoryBrokerMessage struct { + messageId string + body string + deleteReqCh chan struct{} + deleteAckCh chan struct{} + timer *time.Timer +} + +type SimpleInMemoryBroker struct { + visibilityTimeout time.Duration + queue chan *SimpleInMemoryBrokerMessage + inFlights chan *SimpleInMemoryBrokerMessage + expired chan *SimpleInMemoryBrokerMessage + + statics struct { + rwMutex sync.RWMutex + enqueuedMessages int + inFlightMessages int + } +} + +func NewSimpleInMemoryBroker(visibilityTimeout time.Duration) *SimpleInMemoryBroker { + return &SimpleInMemoryBroker{ + visibilityTimeout: visibilityTimeout, + queue: make(chan *SimpleInMemoryBrokerMessage, 1000), + inFlights: make(chan *SimpleInMemoryBrokerMessage), + expired: make(chan *SimpleInMemoryBrokerMessage, 1000), + } +} + +func (b *SimpleInMemoryBroker) run(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case msg := <-b.inFlights: + go func(ctx context.Context) { + select { + case <-ctx.Done(): + return + case <-msg.deleteReqCh: + msg.deleteAckCh <- struct{}{} + case <-msg.timer.C: + b.expired <- msg + } + }(ctx) + } + } +} + +func (b *SimpleInMemoryBroker) AddMessages(msgs []*SimpleInMemoryBrokerMessage) { + for _, msg := range msgs { + b.queue <- msg + b.statics.rwMutex.Lock() + b.statics.enqueuedMessages++ + b.statics.rwMutex.Unlock() + } +} + +func (b *SimpleInMemoryBroker) DeleteMessages(msgs []messages.Message) error { + requestTimer := time.NewTimer(time.Second * 5) + defer requestTimer.Stop() + + for _, msg := range msgs { + brokerMsg := msg.Content().(*SimpleInMemoryBrokerMessage) + + select { + case <-requestTimer.C: + return fmt.Errorf("failed to delete message %s: request timeout", brokerMsg.messageId) + case brokerMsg.deleteReqCh <- struct{}{}: + } + + if !brokerMsg.timer.Stop() { + return fmt.Errorf("failed to delete message %s: visibility timeout exipired", brokerMsg.messageId) + } + + <-brokerMsg.deleteAckCh + + b.statics.rwMutex.Lock() + b.statics.inFlightMessages-- + b.statics.rwMutex.Unlock() + } + + return nil +} + +func (b *SimpleInMemoryBroker) ReceiveMessages(ctx context.Context) ([]messages.Message, error) { + var polledMessage *SimpleInMemoryBrokerMessage + select { + case polledMessage = <-b.expired: + default: + timer := time.NewTimer(time.Millisecond * 500) + defer timer.Stop() + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-timer.C: + return nil, nil + case polledMessage = <-b.expired: + case polledMessage = <-b.queue: + } + } + + polledMessage.timer = time.NewTimer(b.visibilityTimeout) + polledMessage.deleteReqCh = make(chan struct{}) + polledMessage.deleteAckCh = make(chan struct{}) + + time.After(time.Millisecond * 5) + + msg := messages.Message{ + MsgId: polledMessage.messageId, + Msg: polledMessage, + ReceivedTime: time.Now(), + } + + // Set a context with timeout + msg.Ctx, msg.CancelCtx = context.WithTimeout(context.Background(), b.visibilityTimeout) + + // Move the message to inflight + b.inFlights <- polledMessage + b.statics.rwMutex.Lock() + b.statics.enqueuedMessages-- + b.statics.inFlightMessages++ + b.statics.rwMutex.Unlock() + + return []messages.Message{msg}, nil +} + +func (b *SimpleInMemoryBroker) EnqueuedMessages() int { + b.statics.rwMutex.RLock() + defer b.statics.rwMutex.RUnlock() + return b.statics.enqueuedMessages +} + +func (b *SimpleInMemoryBroker) InFlightMessages() int { + b.statics.rwMutex.RLock() + defer b.statics.rwMutex.RUnlock() + return b.statics.inFlightMessages +} + +func TestWorker(t *testing.T) { + inMemoryBroker := NewSimpleInMemoryBroker(time.Second * 10) + go inMemoryBroker.run(context.Background()) + + t.Run("can receive a message", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + msgs := []*SimpleInMemoryBrokerMessage{ + { + messageId: "1", + body: "Hello, world!", + }, + } + + inMemoryBroker.AddMessages(msgs) + + wkr := NewWorker(Configuration{ + Client: inMemoryBroker, + Concurrency: 1, + Retrievers: 1, + ErrorConfig: ErrorConfiguration{ + ReportFunc: func(err error) bool { + t.Fatalf("unexpected error: %v", err) + return true + }, + }, + Consumer: NewMessageConsumer(MessageConsumerConfiguration{ + Handler: func(ctx context.Context, msg Message) error { + defer cancel() + + assert.Equal(t, "Hello, world!", msg.Content().(*SimpleInMemoryBrokerMessage).body) + + return nil + }, + }), + }) + + assert.NoError(t, wkr.Run(ctx)) + assert.Equal(t, 0, inMemoryBroker.EnqueuedMessages()) + assert.Equal(t, 0, inMemoryBroker.InFlightMessages()) + }) + + t.Run("can receive a batch of messages", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + msgs := []*SimpleInMemoryBrokerMessage{ + { + messageId: "1", + body: "Hello, world 1!", + }, + { + messageId: "2", + body: "Hello, world 2!", + }, + { + messageId: "3", + body: "Hello, world 3!", + }, + } + + inMemoryBroker.AddMessages(msgs) + + wkr := NewWorker(Configuration{ + Client: inMemoryBroker, + Concurrency: 1, + Retrievers: 1, + ErrorConfig: ErrorConfiguration{ + ReportFunc: func(err error) bool { + t.Fatalf("unexpected error: %v", err) + return true + }, + }, + Consumer: NewBatchConsumer(BatchConsumerConfiguration{ + BufferConfig: BatchConsumerBufferConfiguration{ + Size: 3, + Timeout: time.Second, + }, + Handler: func(ctx context.Context, msgs []Message) (BatchResponse, error) { + defer cancel() + + if len(msgs) < 3 { + t.Fatalf("expected 3 messages, got %d", len(msgs)) + } + + assert.Equal(t, "Hello, world 1!", msgs[0].Content().(*SimpleInMemoryBrokerMessage).body) + assert.Equal(t, "Hello, world 2!", msgs[1].Content().(*SimpleInMemoryBrokerMessage).body) + assert.Equal(t, "Hello, world 3!", msgs[2].Content().(*SimpleInMemoryBrokerMessage).body) + + return BatchResponse{}, nil + }, + }), + }) + + assert.NoError(t, wkr.Run(ctx)) + assert.Equal(t, 0, inMemoryBroker.EnqueuedMessages()) + assert.Equal(t, 0, inMemoryBroker.InFlightMessages()) + }) +} From 740ffa77725f44579d7350d54d8d989f955f3f38 Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Tue, 29 Sep 2026 11:48:40 +0100 Subject: [PATCH 02/11] feat(PAR-2054): remove halting for 'too many errors' entirely --- README.md | 2 +- v2/config.go | 35 ++++++++------------------------ v2/consumers.go | 4 ++-- v2/controller.go | 19 ++++++----------- v2/internal/messages/messages.go | 8 ++++---- v2/message.go | 4 ++-- v2/worker.go | 12 ++++------- v2/worker_test.go | 14 ++++--------- 8 files changed, 31 insertions(+), 67 deletions(-) diff --git a/README.md b/README.md index e7ff1b1..f40ba50 100644 --- a/README.md +++ b/README.md @@ -152,7 +152,7 @@ func main() { // Collect the ids of any messages that failed. They won't be deleted, // so they return to the queue once their visibility timeout expires. - var failed []interface{} + var failed []any for _, msg := range msgs { // Assert the type of message to get the body or any other attributes diff --git a/v2/config.go b/v2/config.go index 18e254d..351f144 100644 --- a/v2/config.go +++ b/v2/config.go @@ -8,8 +8,6 @@ import ( ) const ( - defaultErrorThreshold = 3 - defaultErrorPeriod = time.Second * 120 defaultConcurrency = 100 defaultRetrievers = 1 defaultDeleterBufferSize = 10 @@ -21,22 +19,6 @@ type DeleterConfiguration struct { BufferTimeout time.Duration } -// The ErrorConfiguration defines a threshold for which the worker stops. If the number -// of errors occurred during the worker execution passes the given Threshold on the -// specified Period, the worker stops. -type ErrorConfiguration struct { - // Number of errors that must occur in the Period before the worker stops. - // Default: 3. - Threshold int - - // Duration of the period for which, if the number of errors passes the Threshold, the worker stops. - // Default: 120s. - Period time.Duration - - // The error report function, returns a boolean value to decide whether the error counts towards to threshold - ReportFunc func(err error) -} - // The BatchConsumerBufferConfiguration defines a buffer which is consumed by the worker when either // the buffer is full or the timeout has passed since the first message got added. type BatchConsumerBufferConfiguration struct { @@ -62,16 +44,16 @@ type BatchConsumerConfiguration struct { } type Configuration struct { - // A queue client + // Client is a queue client. Client client.Client - // Number of Go routines that process the messages from the Queue. + // Concurrency is the number of Go routines that process the messages from the Queue. // The higher this value, the more Go routines are spawned to process the messages. // Using a high value can be useful when the Handler of the consumer perform slow I/O operations. // Default: 100. Concurrency int - // Number of Go routines that retrieve messages from the Queue. + // Retrievers is the number of Go routines that retrieve messages from the Queue. // The higher this value, the more Go routines are spawned to read the messages from the // queue and provide them to the worker's consumers. // Using a high value can be useful when the network is slow or when consumers are quicker @@ -79,8 +61,9 @@ type Configuration struct { // Default: 1. Retrievers int - // The ErrorConfiguration. - ErrorConfig ErrorConfiguration + // ReportFunc will log/report an error as needed by the user. + // Default: logs errors. Set to an empty func to disable. + ReportFunc func(error) // The messages Consumer. Consumer Consumer @@ -98,11 +81,9 @@ func setWorkerConfigValues(config Configuration) Configuration { config.Concurrency = defaultConcurrency } - if config.ErrorConfig.ReportFunc == nil { - config.ErrorConfig.ReportFunc = func(err error) bool { + if config.ReportFunc == nil { + config.ReportFunc = func(err error) { log.Println("ERROR", err) - - return true } } diff --git a/v2/consumers.go b/v2/consumers.go index e45ddb5..59d619c 100644 --- a/v2/consumers.go +++ b/v2/consumers.go @@ -11,7 +11,7 @@ import ( ) type BatchResponse struct { - FailedMessagesId []interface{} + FailedMessagesId []any } type MessageHandler = func(ctx context.Context, msg Message) error @@ -225,7 +225,7 @@ func (c *batchConsumer) buildMessagesToDeleteFromBatchResponse(msgs []messages.M toDelete := make([]messages.Message, 0, len(msgs)) - failedMessagesIdIndexed := make(map[interface{}]struct{}, len(resp.FailedMessagesId)) + failedMessagesIdIndexed := make(map[any]struct{}, len(resp.FailedMessagesId)) for _, id := range resp.FailedMessagesId { failedMessagesIdIndexed[id] = struct{}{} } diff --git a/v2/controller.go b/v2/controller.go index 429120b..d24ec75 100644 --- a/v2/controller.go +++ b/v2/controller.go @@ -1,24 +1,17 @@ package formigo -import ( - "context" - "sync" -) - type controller struct { - errorConfig ErrorConfiguration - errorCounter int - mutex sync.Mutex - stopOnce sync.Once - stopFunc context.CancelCauseFunc + reportFunc func(error) } func (c *controller) reportError(err error) { - c.errorConfig.ReportFunc(err) + if c.reportFunc != nil { + c.reportFunc(err) + } } -func newController(errorConfig ErrorConfiguration) *controller { +func newController(reportFunc func(error)) *controller { return &controller{ - errorConfig: errorConfig, + reportFunc: reportFunc, } } diff --git a/v2/internal/messages/messages.go b/v2/internal/messages/messages.go index ba91502..c3a7c67 100644 --- a/v2/internal/messages/messages.go +++ b/v2/internal/messages/messages.go @@ -8,16 +8,16 @@ import ( type Message struct { Ctx context.Context `json:"-"` // Exclude from JSON CancelCtx context.CancelFunc `json:"-"` // Exclude from JSON - MsgId interface{} `json:"id"` - Msg interface{} `json:"content"` + MsgId any `json:"id"` + Msg any `json:"content"` ReceivedTime time.Time `json:"receivedAt"` } -func (m Message) Id() interface{} { +func (m Message) Id() any { return m.MsgId } -func (m Message) Content() interface{} { +func (m Message) Content() any { return m.Msg } diff --git a/v2/message.go b/v2/message.go index d1b118a..dbf5278 100644 --- a/v2/message.go +++ b/v2/message.go @@ -4,6 +4,6 @@ import "time" type Message interface { ReceivedAt() time.Time - Content() interface{} - Id() interface{} + Content() any + Id() any } diff --git a/v2/worker.go b/v2/worker.go index 654a732..b0fe2d6 100644 --- a/v2/worker.go +++ b/v2/worker.go @@ -13,18 +13,14 @@ type Worker struct { client client.Client concurrency int retrievers int - errorConfig ErrorConfiguration + reportFunc func(error) consumer Consumer deleterConfig DeleterConfiguration } func (w Worker) Run(ctx context.Context) error { - // Create a new context with a cancel function used to stop the worker from the - // controller in case too many errors occur. - ctx, cancel := context.WithCancelCause(ctx) - - // Create controller - ctrl := newController(w.errorConfig, cancel) + // Create a controller + ctrl := newController(w.reportFunc) // Run retrievers and get the message channel messageCh := w.runRetrievers(ctx, ctrl) @@ -103,7 +99,7 @@ func NewWorker(config Configuration) Worker { client: config.Client, concurrency: config.Concurrency, retrievers: config.Retrievers, - errorConfig: config.ErrorConfig, + reportFunc: config.ReportFunc, consumer: config.Consumer, deleterConfig: config.DeleterConfig, } diff --git a/v2/worker_test.go b/v2/worker_test.go index e00bada..823b17f 100644 --- a/v2/worker_test.go +++ b/v2/worker_test.go @@ -172,11 +172,8 @@ func TestWorker(t *testing.T) { Client: inMemoryBroker, Concurrency: 1, Retrievers: 1, - ErrorConfig: ErrorConfiguration{ - ReportFunc: func(err error) bool { - t.Fatalf("unexpected error: %v", err) - return true - }, + ReportFunc: func(err error) { + t.Fatalf("unexpected error: %v", err) }, Consumer: NewMessageConsumer(MessageConsumerConfiguration{ Handler: func(ctx context.Context, msg Message) error { @@ -217,11 +214,8 @@ func TestWorker(t *testing.T) { Client: inMemoryBroker, Concurrency: 1, Retrievers: 1, - ErrorConfig: ErrorConfiguration{ - ReportFunc: func(err error) bool { - t.Fatalf("unexpected error: %v", err) - return true - }, + ReportFunc: func(err error) { + t.Fatalf("unexpected error: %v", err) }, Consumer: NewBatchConsumer(BatchConsumerConfiguration{ BufferConfig: BatchConsumerBufferConfiguration{ From daf2976765d778cdfad50144d0cc36d10f65df83 Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Tue, 29 Sep 2026 18:11:03 +0100 Subject: [PATCH 03/11] feat(PAR-2054): fully rework formigo to use pools --- v2/README.md | 12 ++ v2/config.go | 99 ------------ v2/consumers.go | 261 ------------------------------- v2/controller.go | 17 -- v2/deleter.go | 63 -------- v2/go.mod | 1 + v2/go.sum | 2 + v2/internal/client/client.go | 20 --- v2/internal/messages/messages.go | 191 ---------------------- v2/manager.go | 124 +++++++++++++++ v2/message.go | 19 ++- v2/retriever.go | 58 ------- v2/sqs.go | 164 ++++++++----------- v2/worker.go | 106 ------------- v2/worker_test.go | 245 ----------------------------- 15 files changed, 215 insertions(+), 1167 deletions(-) create mode 100644 v2/README.md delete mode 100644 v2/config.go delete mode 100644 v2/consumers.go delete mode 100644 v2/controller.go delete mode 100644 v2/deleter.go delete mode 100644 v2/internal/client/client.go delete mode 100644 v2/internal/messages/messages.go create mode 100644 v2/manager.go delete mode 100644 v2/retriever.go delete mode 100644 v2/worker.go delete mode 100644 v2/worker_test.go diff --git a/v2/README.md b/v2/README.md new file mode 100644 index 0000000..3169cad --- /dev/null +++ b/v2/README.md @@ -0,0 +1,12 @@ +# Formigo - distributed SQS worker pools. + +> This library relies heavily on [`github.com/alitto/pond/v2`](https://github.com/alitto/pond), a very flexible and well-tested worker pool library. + +Formigo is a fast, reliable worker pool consumer for SQS. + +Basic features of Pond: + +- automatic scaling to available resources/limits based on incoming queue pressure. +- fire & forget queue submission +- fire & **wait for a response** submission to a queue (e.g. for dependent jobs, or deferred behaviour). +- clean, graceful exit when shutting down. \ No newline at end of file diff --git a/v2/config.go b/v2/config.go deleted file mode 100644 index 351f144..0000000 --- a/v2/config.go +++ /dev/null @@ -1,99 +0,0 @@ -package formigo - -import ( - "log" - "time" - - "github.com/Pod-Point/go-queue-worker/v2/internal/client" -) - -const ( - defaultConcurrency = 100 - defaultRetrievers = 1 - defaultDeleterBufferSize = 10 - defaultDeleterBufferTimeout = time.Millisecond * 500 -) - -type DeleterConfiguration struct { - BufferSize int - BufferTimeout time.Duration -} - -// The BatchConsumerBufferConfiguration defines a buffer which is consumed by the worker when either -// the buffer is full or the timeout has passed since the first message got added. -type BatchConsumerBufferConfiguration struct { - // Max number of messages that the buffer can contain. - // Default: 10. - Size int - - // Time after which the buffer gets processed, no matter whether it is full or not. - // This value MUST be smaller tha VisibilityTimeout in the - // RetrieveMessageConfiguration + the maximum processing time of the handler. - // If this is not set correctly, the same message could be processed multiple times. - // Default: 1s. - Timeout time.Duration -} - -type MessageConsumerConfiguration struct { - Handler MessageHandler -} - -type BatchConsumerConfiguration struct { - Handler BatchHandler - BufferConfig BatchConsumerBufferConfiguration -} - -type Configuration struct { - // Client is a queue client. - Client client.Client - - // Concurrency is the number of Go routines that process the messages from the Queue. - // The higher this value, the more Go routines are spawned to process the messages. - // Using a high value can be useful when the Handler of the consumer perform slow I/O operations. - // Default: 100. - Concurrency int - - // Retrievers is the number of Go routines that retrieve messages from the Queue. - // The higher this value, the more Go routines are spawned to read the messages from the - // queue and provide them to the worker's consumers. - // Using a high value can be useful when the network is slow or when consumers are quicker - // than retrievers. - // Default: 1. - Retrievers int - - // ReportFunc will log/report an error as needed by the user. - // Default: logs errors. Set to an empty func to disable. - ReportFunc func(error) - - // The messages Consumer. - Consumer Consumer - - // Configuration for the deleter - DeleterConfig DeleterConfiguration -} - -func setWorkerConfigValues(config Configuration) Configuration { - if config.Retrievers == 0 { - config.Retrievers = defaultRetrievers - } - - if config.Concurrency == 0 { - config.Concurrency = defaultConcurrency - } - - if config.ReportFunc == nil { - config.ReportFunc = func(err error) { - log.Println("ERROR", err) - } - } - - if config.DeleterConfig.BufferSize == 0 { - config.DeleterConfig.BufferSize = defaultDeleterBufferSize - } - - if config.DeleterConfig.BufferTimeout == 0 { - config.DeleterConfig.BufferTimeout = defaultDeleterBufferTimeout - } - - return config -} diff --git a/v2/consumers.go b/v2/consumers.go deleted file mode 100644 index 59d619c..0000000 --- a/v2/consumers.go +++ /dev/null @@ -1,261 +0,0 @@ -package formigo - -import ( - "context" - "errors" - "fmt" - "sync" - "time" - - "github.com/Pod-Point/go-queue-worker/v2/internal/messages" -) - -type BatchResponse struct { - FailedMessagesId []any -} - -type MessageHandler = func(ctx context.Context, msg Message) error -type BatchHandler = func(ctx context.Context, msgs []Message) (BatchResponse, error) - -// This means that the buffered messages didn't get passed to the handler within -// the first message's timeout. -// This is generally due to: -// - Visibility timeout of the messages too small -// - Buffer timeout too high -// - Consumer to slow -var errBufferCtxExpired = errors.New("buffer context expired, buffer will Reset") - -type Consumer interface { - consume(concurrency int, ctrl *controller, messageCh <-chan messages.Message, deleteCh chan<- messages.Message) -} - -func makeAvailableConsumers(concurrency int) chan struct{} { - consumers := make(chan struct{}, concurrency) - for i := 0; i < concurrency; i++ { - consumers <- struct{}{} - } - - return consumers -} - -// wrapHandler catches any panic error and returns the error that generated it. -// It prevents the worker from crashing in case of an unexpected error. -func wrapHandler(handler func() error) (err error) { - defer func() { - if r := recover(); r != nil { - // Set error before returning - err = fmt.Errorf("panic error: %s", r) - } - }() - - err = handler() - - return err -} - -// messageConsumer defines a message handler that consumes only one message at a -// time. -// It can be useful when the workload is specific per message, for example for sending -// an email. -type messageConsumer struct { - handler MessageHandler -} - -func (c *messageConsumer) processMessage(msg messages.Message) error { - defer msg.CancelCtx() // This must be called to release resources associated with the context. - - // Process Message - return wrapHandler(func() error { - return c.handler(msg.Ctx, msg) - }) -} - -// Consumes and deletes a single message, it stops only when the `messageCh` gets closed -// and doesn't have any messages in it. -func (c *messageConsumer) consume(concurrency int, ctrl *controller, messageCh <-chan messages.Message, deleteCh chan<- messages.Message) { - consumers := makeAvailableConsumers(concurrency) - - var wg sync.WaitGroup - for msg := range messageCh { - <-consumers // Use an available consumer - - wg.Add(1) - go func(message messages.Message) { - defer func() { - wg.Done() - consumers <- struct{}{} // Release consumer - }() - - err := c.processMessage(message) - if err != nil { - ctrl.reportError(fmt.Errorf("failed to process message: %w", err)) - return - } - - // Push message for deletion - deleteCh <- message - }(msg) - } - - wg.Wait() -} - -func NewMessageConsumer(config MessageConsumerConfiguration) *messageConsumer { - return &messageConsumer{ - handler: config.Handler, - } -} - -// batchConsumer allows to process multiple messages at a time. This can be useful -// for batch updates or use cases with high throughput. -type batchConsumer struct { - handler BatchHandler - bufferConfig BatchConsumerBufferConfiguration -} - -// It processes the messages and push them downstream for deletion. -func (c *batchConsumer) processMessages(ctrl *controller, deleteCh chan<- messages.Message, ctx context.Context, msgs []messages.Message) { - defer func() { - if r := recover(); r != nil { - ctrl.reportError(fmt.Errorf("panic error: %s", r)) - } - }() - - // Convert slice to the abstraction - converted := make([]Message, 0, len(msgs)) - for _, msg := range msgs { - converted = append(converted, msg) - } - - resp, err := c.handler(ctx, converted) - if err != nil { - ctrl.reportError(fmt.Errorf("failed to process batch: %w", err)) - } - - toDelete := c.buildMessagesToDeleteFromBatchResponse(msgs, resp) - // Push messages for deletion - for _, msg := range toDelete { - deleteCh <- msg - } -} - -// Consumes and deletes a number of messages in the interval [1, N] based on configuration -// provided in the BufferConfiguration. -// It stops only when the messageCh gets closed and doesn't have any messages in it. -func (c *batchConsumer) consume(concurrency int, ctrl *controller, messageCh <-chan messages.Message, deleteCh chan<- messages.Message) { - consumers := makeAvailableConsumers(concurrency) - - // Create buffer - buffer := messages.NewBufferWithContextTimeout(messages.BufferWithContextTimeoutConfiguration{ - Size: c.bufferConfig.Size, - BufferTimeout: c.bufferConfig.Timeout, - }) - defer buffer.Reset() - - var wg sync.WaitGroup - func() { - for { - select { - case msg, open := <-messageCh: - if !open { - // Message channel closed. This is the stop signal. - // Note: We can't return if the buffer contains messages to process. - // We MUST consume all the messages on the buffer - if buffer.IsEmpty() { - return // Buffer empty, we can stop - } - break // Buffer contains messages, break the select - } - - // Add message to the buffer - buffer.Add(msg) - - // If the buffer is not full, continue - if !buffer.IsFull() { - continue - } - - case <-buffer.CtxExpired(): - ctrl.reportError(errBufferCtxExpired) - - // Reset the buffer. - buffer.Reset() - continue - - case <-buffer.Expired(): - // Timeout expired, process the buffer - } - - select { - case <-consumers: // Use an available consumer - case <-buffer.CtxExpired(): - ctrl.reportError(errBufferCtxExpired) - - // Reset the buffer. - buffer.Reset() - continue - } - - ctx, cancelCtx := buffer.PullContext() - - wg.Add(1) - go func(ctx context.Context, ctxCancelFunc context.CancelFunc, msgs []messages.Message) { - defer func() { - wg.Done() - consumers <- struct{}{} // Release consumer - ctxCancelFunc() // Cancel context - }() - - // Process the messages - c.processMessages(ctrl, deleteCh, ctx, msgs) - }(ctx, cancelCtx, buffer.Messages()) - - // Reset buffer - buffer.Reset() - } - }() - - wg.Wait() -} - -func (c *batchConsumer) buildMessagesToDeleteFromBatchResponse(msgs []messages.Message, resp BatchResponse) []messages.Message { - if len(resp.FailedMessagesId) == 0 { - return msgs - } - - toDelete := make([]messages.Message, 0, len(msgs)) - - failedMessagesIdIndexed := make(map[any]struct{}, len(resp.FailedMessagesId)) - for _, id := range resp.FailedMessagesId { - failedMessagesIdIndexed[id] = struct{}{} - } - - for _, msg := range msgs { - if _, ok := failedMessagesIdIndexed[msg.Id()]; !ok { - toDelete = append(toDelete, msg) - } - } - - return toDelete -} - -func NewBatchConsumer(config BatchConsumerConfiguration) *batchConsumer { - if config.BufferConfig.Size == 0 { - config.BufferConfig.Size = 10 - } - - if config.BufferConfig.Timeout == 0 { - config.BufferConfig.Timeout = time.Second - } - - return &batchConsumer{ - handler: config.Handler, - bufferConfig: config.BufferConfig, - } -} - -// Interface guards -var ( - _ Consumer = (*messageConsumer)(nil) - _ Consumer = (*batchConsumer)(nil) -) diff --git a/v2/controller.go b/v2/controller.go deleted file mode 100644 index d24ec75..0000000 --- a/v2/controller.go +++ /dev/null @@ -1,17 +0,0 @@ -package formigo - -type controller struct { - reportFunc func(error) -} - -func (c *controller) reportError(err error) { - if c.reportFunc != nil { - c.reportFunc(err) - } -} - -func newController(reportFunc func(error)) *controller { - return &controller{ - reportFunc: reportFunc, - } -} diff --git a/v2/deleter.go b/v2/deleter.go deleted file mode 100644 index f392ff0..0000000 --- a/v2/deleter.go +++ /dev/null @@ -1,63 +0,0 @@ -package formigo - -import ( - "fmt" - "sync" - - "github.com/Pod-Point/go-queue-worker/v2/internal/client" - "github.com/Pod-Point/go-queue-worker/v2/internal/messages" -) - -// deleter will delete messages from SQS until the delete channel gets closed. -// Any error will be sent to the error channel. -func deleter(wg *sync.WaitGroup, deleter client.MessageDeleter, config DeleterConfiguration, ctrl *controller, deleteCh <-chan messages.Message) { - // Create buffer - buffer := messages.NewMessageBuffer(messages.BufferConfiguration{ - Size: config.BufferSize, - Timeout: config.BufferTimeout, - }) - defer buffer.Reset() - - for { - select { - case msg, open := <-deleteCh: - if !open { - // Delete channel closed. This is the stop signal. - // Note: We can't return if the buffer contains messages to delete. - // We MUST send the deletion request if the buffer has any messages - if buffer.IsEmpty() { - return // Buffer empty, we can stop - } - break // Buffer contains messages, break the select - } - - buffer.Add(msg) - - // If the buffer is not full, continue - if !buffer.IsFull() { - continue - } - - case <-buffer.Expired(): - // Buffer expired, process the buffer - } - - // Get the messages - msgs := buffer.Messages() - - // Send deletion request in a separate Go routine to maximise the throughput - wg.Add(1) - go func(msgs []messages.Message) { - defer wg.Done() - // No context should be passed here. In fact, if the messages were processed correctly - // we want to do our best to delete it from the queue. - err := deleter.DeleteMessages(msgs) - if err != nil { - ctrl.reportError(fmt.Errorf("unable to delete %d messages: %w", len(msgs), err)) - } - }(msgs) - - // Reset buffer - buffer.Reset() - } -} diff --git a/v2/go.mod b/v2/go.mod index cee5512..2d49cc2 100644 --- a/v2/go.mod +++ b/v2/go.mod @@ -8,6 +8,7 @@ require ( ) require ( + github.com/alitto/pond/v2 v2.7.1 // indirect github.com/aws/aws-sdk-go-v2 v1.30.4 // indirect github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.16 // indirect github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.16 // indirect diff --git a/v2/go.sum b/v2/go.sum index 5c94707..bc591a3 100644 --- a/v2/go.sum +++ b/v2/go.sum @@ -1,3 +1,5 @@ +github.com/alitto/pond/v2 v2.7.1 h1:QxMbcfjcVTa0pyxX5Ib1226mM8u8D7gKUVkCUU4DYIw= +github.com/alitto/pond/v2 v2.7.1/go.mod h1:xkjYEgQ05RSpWdfSd1nM3OVv7TBhLdy7rMp3+2Nq+yE= github.com/aws/aws-sdk-go-v2 v1.30.4 h1:frhcagrVNrzmT95RJImMHgabt99vkXGslubDaDagTk8= github.com/aws/aws-sdk-go-v2 v1.30.4/go.mod h1:CT+ZPWXbYrci8chcARI3OmI/qgd+f6WtuLOoaIA8PR0= github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.16 h1:TNyt/+X43KJ9IJJMjKfa3bNTiZbUP7DeCxfbTROESwY= diff --git a/v2/internal/client/client.go b/v2/internal/client/client.go deleted file mode 100644 index 0af07a8..0000000 --- a/v2/internal/client/client.go +++ /dev/null @@ -1,20 +0,0 @@ -package client - -import ( - "context" - - "github.com/Pod-Point/go-queue-worker/v2/internal/messages" -) - -type MessageReceiver interface { - ReceiveMessages(ctx context.Context) ([]messages.Message, error) -} - -type MessageDeleter interface { - DeleteMessages(messages []messages.Message) error -} - -type Client interface { - MessageReceiver - MessageDeleter -} diff --git a/v2/internal/messages/messages.go b/v2/internal/messages/messages.go deleted file mode 100644 index c3a7c67..0000000 --- a/v2/internal/messages/messages.go +++ /dev/null @@ -1,191 +0,0 @@ -package messages - -import ( - "context" - "time" -) - -type Message struct { - Ctx context.Context `json:"-"` // Exclude from JSON - CancelCtx context.CancelFunc `json:"-"` // Exclude from JSON - MsgId any `json:"id"` - Msg any `json:"content"` - ReceivedTime time.Time `json:"receivedAt"` -} - -func (m Message) Id() any { - return m.MsgId -} - -func (m Message) Content() any { - return m.Msg -} - -func (m Message) ReceivedAt() time.Time { - return m.ReceivedTime -} - -type BufferConfiguration struct { - Size int - Timeout time.Duration -} - -// Buffer is used to implement a buffer with a size and timeout. -// When the buffer is full, the `full()` method returns true. -// If the buffer has expired, the `timer` will emit. -// Note that buffer is not thread safe. Remember to use a mutex -// in case it's been used by multiple Go routines. -type Buffer struct { - messages []Message - - // expiredCh emits when the buffer expires. - // When the buffer is initialised or reset the channel is set - // so that it never expires. - expiredCh <-chan time.Time - - size int - timeout time.Duration - timer *time.Timer -} - -// Len returns the number of Messages in the buffer. -func (b *Buffer) Len() int { - return len(b.messages) -} - -// IsFull Returns true if the buffer is full. -func (b *Buffer) IsFull() bool { - return len(b.messages) == b.size -} - -// Add a message to the buffer and set the timer if the buffer was empty. -func (b *Buffer) Add(msg Message) { - if len(b.messages) == 0 { - // Create a new timer and assign its channel to expiredCh - b.timer = time.NewTimer(b.timeout) - b.expiredCh = b.timer.C - } - - b.messages = append(b.messages, msg) -} - -// IsEmpty returns true if the buffer is empty. -func (b *Buffer) IsEmpty() bool { - return len(b.messages) == 0 -} - -// Messages returns the messages held in the buffer. -func (b *Buffer) Messages() []Message { - return b.messages -} - -// Reset flush the messages contained in the buffer and stop the timer. -// It's important to stop the timer to avoid memory leaks. In fact, the -// GC won't collect the timer until its channel expires. -// NOTE: this function should be always called to clean up any buffer -// created. Used in defer can guarantee that it always run. -func (b *Buffer) Reset() { - if b.timer != nil { - // Stop the timer to free its resources - b.timer.Stop() - b.timer = nil - - // Reset the expiredCh so that it never expires - b.expiredCh = make(<-chan time.Time) - } - - if len(b.messages) > 0 { - b.messages = make([]Message, 0, b.size) - } -} - -// Expired emits when the buffer expires. -func (b *Buffer) Expired() <-chan time.Time { - return b.expiredCh -} - -func NewMessageBuffer(config BufferConfiguration) *Buffer { - return &Buffer{ - expiredCh: make(<-chan time.Time), - messages: make([]Message, 0, config.Size), - size: config.Size, - timeout: config.Timeout, - timer: nil, - } -} - -type BufferWithContextTimeoutConfiguration struct { - BufferTimeout time.Duration - CtxTimeout time.Duration - Size int -} - -// BufferWithContextTimeout is used to construct a buffer that has a context timeout -// along with the standard buffer timeout. This is used because the messages have to -// be processed within a certain period and if this doesn't happen, the buffer should -// delete the messages in it and reset. -type BufferWithContextTimeout struct { - *Buffer - ctx context.Context - cancelCtx context.CancelFunc -} - -func (b *BufferWithContextTimeout) Add(msg Message) { - if len(b.messages) == 0 { - // Set the context of the buffer to first message's context - b.ctx = msg.Ctx - } - - // Override the current cancelCtx in a way that cancels all - // the previous messages' contexts. - b.cancelCtx = func(cancel context.CancelFunc) context.CancelFunc { - return func() { - cancel() - msg.CancelCtx() - } - }(b.cancelCtx) - - b.Buffer.Add(msg) -} - -// Reset resets its internal buffer, cancel the current context created and -// reset any timeout. -// It's important to call this function to avoid memory leaks. In fact, the -// GC won't collect any timer or resources allocated within the context. -// NOTE: this function should be always called to clean up any buffer -// created. Used in defer can guarantee that it always run. -func (b *BufferWithContextTimeout) Reset() { - b.Buffer.Reset() - - b.cancelCtx() // Be sure to reset any previous context - b.ctx = context.Background() // Create a context that doesn't expire - b.cancelCtx = func() {} -} - -func (b *BufferWithContextTimeout) CtxExpired() <-chan struct{} { - return b.ctx.Done() -} - -func (b *BufferWithContextTimeout) Context() context.Context { - return b.ctx -} - -func (b *BufferWithContextTimeout) PullContext() (context.Context, context.CancelFunc) { - ctx, cancelCtx := b.ctx, b.cancelCtx - - b.ctx = context.Background() // Create a context that doesn't expire - b.cancelCtx = func() {} - - return ctx, cancelCtx -} - -func NewBufferWithContextTimeout(config BufferWithContextTimeoutConfiguration) *BufferWithContextTimeout { - return &BufferWithContextTimeout{ - Buffer: NewMessageBuffer(BufferConfiguration{ - Size: config.Size, - Timeout: config.BufferTimeout, - }), - ctx: context.Background(), - cancelCtx: func() {}, // Empty cancelCtx - } -} diff --git a/v2/manager.go b/v2/manager.go new file mode 100644 index 0000000..ca1deab --- /dev/null +++ b/v2/manager.go @@ -0,0 +1,124 @@ +package formigo + +import ( + "context" + "time" + + "github.com/alitto/pond/v2" +) + +// Fetcher condenses RetrieveMessage into the smallest possible interface. +type Fetcher interface { + Fetch(context.Context) ([]Message, error) +} + +// Deleter condenses DeleteMessage into the smallest possible interface. +type Deleter interface { + Delete(context.Context, Message) error +} + +type Client interface { + Fetcher + Deleter +} + +// Manager sets up a set of workers and processes an queue via them. +type Manager struct { + client Client + pool pond.Pool + deadline time.Duration + fetchDelay time.Duration + + deleter func(context.Context, Message) error + consumer func(context.Context, Message) error + reporter func(error) +} + +// Submit will push the message onto a consumer via the worker pool. +func (m *Manager) Submit(msg Message) error { + m.pool.SubmitErr(func() error { + ctx := context.Background() + + // make sure this consumer deletes the message from the queue when deferring. + // this one uses the background context with no deadline, and reports its own errors. + defer m.Delete(ctx, msg) + + if m.deadline > 0 { + var cancel func() + ctx, cancel = context.WithDeadline(ctx, time.Now().Add(m.deadline)) + defer cancel() + } + + if err := m.consumer(ctx, msg); err != nil { + m.reporter(err) + return err + } + + return nil + }) + + return nil +} + +// Delete will call the Client's Delete method, skipping any batch deletes. +func (m *Manager) Delete(ctx context.Context, msg Message) { + if err := m.client.Delete(ctx, msg); err != nil { + m.reporter(err) + } +} + +func (m *Manager) Fetch(ctx context.Context) error { + for { + select { + case <-ctx.Done(): + default: + messages, err := m.client.Fetch(ctx) + if err != nil { + m.Report(err) // report errors with the initial fetch. + continue + } + + for _, msg := range messages { + if err := m.Submit(msg); err != nil { + m.Report(err) + } + } + + // allow configuring a delay after fetching messages + // this may help prevent excessive pressure on AWS. + time.Sleep(m.fetchDelay) + } + } +} + +// Report will report an error whenever a func for it is set. +func (m *Manager) Report(err error) { + if m.reporter != nil { + m.reporter(err) + } +} + +// Run will start up a worker pool and start processing a queue. +// It must exit via the context, e.g. via signal.NotifyContext. +// Using context.Background will never safely exit. +func (m *Manager) Run(ctx context.Context) error { + m.pool = pond.NewPool(20, pond.WithContext(ctx)) + + // set up the fetchers to feed the workers and start them + fetcher := pond.NewPool(2, pond.WithContext(ctx)) + for range 2 { + fetcher.SubmitErr(func() error { + return m.Fetch(ctx) + }) + } + + select { + case <-ctx.Done(): + // stop the fetcher first to be sure it has drained + fetcher.StopAndWait() + + // stop and drain the worker pool + m.pool.StopAndWait() + return ctx.Err() + } +} diff --git a/v2/message.go b/v2/message.go index dbf5278..e38d92e 100644 --- a/v2/message.go +++ b/v2/message.go @@ -1,9 +1,18 @@ package formigo -import "time" +import ( + "encoding/json" +) -type Message interface { - ReceivedAt() time.Time - Content() any - Id() any +type Message struct { + ID string + Attributes map[string]string + Body string + MessageAttributes map[string]any + ReceiptHandle string +} + +// Decode attempts to decode a message's Body using json.Unmarshal. +func (m Message) Decode(value any) error { + return json.Unmarshal([]byte(m.Body), &value) } diff --git a/v2/retriever.go b/v2/retriever.go deleted file mode 100644 index e3da536..0000000 --- a/v2/retriever.go +++ /dev/null @@ -1,58 +0,0 @@ -package formigo - -import ( - "context" - "errors" - "fmt" - - "github.com/Pod-Point/go-queue-worker/v2/internal/client" - "github.com/Pod-Point/go-queue-worker/v2/internal/messages" -) - -// retriever will get messages from SQS until the given context gets canceled. -// Any error will be sent to the controller. -func retriever(ctx context.Context, receiver client.MessageReceiver, ctrl *controller, messageCh chan<- messages.Message) { - for { - select { - case <-ctx.Done(): - return - default: - msgs, err := receiver.ReceiveMessages(ctx) - if err != nil { - if errors.Is(err, context.Canceled) && errors.Is(ctx.Err(), context.Canceled) { - // The worker's context was canceled. We can exit. - return - } - - // Report the error to the controller and continue. - ctrl.reportError(fmt.Errorf("unable to receive message: %w", err)) - continue - } - - // All the messages retrieved must be processed. - // This means that the retriever won't listen for context cancellation - // at this stage. - func() { - for _, msg := range msgs { - select { - case <-msg.Ctx.Done(): - // If consumers don't pick up the messages within the messages' timeout we raise - // an error. - // This could be due to one or more of the following reasons: - // - message timeout too small. - // - consumer too slow. Increasing the number of consumers may help, especially if - // the handler performs many I/O operations. - // - // Note that we won't process all messages retrieved by the API calls. This is because - // the visibility timeout is the same for all the messages returned by the call. - ctrl.reportError(errors.New("message didn't get picked up by any consumer within its timeout")) - - return // Avoid publishing all the messages downstream - case messageCh <- msg: - // Message pushed to the channel - } - } - }() - } - } -} diff --git a/v2/sqs.go b/v2/sqs.go index 268fcf4..71211ae 100644 --- a/v2/sqs.go +++ b/v2/sqs.go @@ -2,138 +2,98 @@ package formigo import ( "context" - "fmt" - "strconv" - "time" + "errors" - awsSqs "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/aws/aws-sdk-go-v2/service/sqs" "github.com/aws/aws-sdk-go-v2/service/sqs/types" - - "github.com/Pod-Point/go-queue-worker/v2/internal/client" - "github.com/Pod-Point/go-queue-worker/v2/internal/messages" ) -type SqsClientConfiguration struct { - // The AWS Sqs Service Client - Svc *awsSqs.Client - - // The AWS ReceiveMessageInput - ReceiveMessageInput *awsSqs.ReceiveMessageInput +var ( + ErrClientNil = errors.New("formigo/v2: sqs client is nil") + ErrInputNil = errors.New("formigo/v2: sqs ReceiveMessage input is nil") + ErrReceiptHandleEmpty = errors.New("formigo/v2: message ReceiptHandle is empty when deleting") +) - // Defines the interval within which the message must be processed. - // If empty, it tries to set the value from the ReceiveMessageInput's - // VisibilityTimeout. - // If ReceiveMessageInput's VisibilityTimeout is empty it retrieves the - // default value set on the queue. This action will fail if the client - // does not have the permission to retrieve the SQS queue's attributes. - MessageCtxTimeout time.Duration +type SQSClient struct { + client *sqs.Client + input *sqs.ReceiveMessageInput } -type sqsClient struct { - svc *awsSqs.Client - receiveMessageInput *awsSqs.ReceiveMessageInput - messageCtxTimeout time.Duration -} +func (s SQSClient) Fetch(ctx context.Context) ([]Message, error) { + if s.client == nil { + return nil, ErrClientNil + } -func (c sqsClient) ReceiveMessages(ctx context.Context) ([]messages.Message, error) { - out, err := c.svc.ReceiveMessage(ctx, c.receiveMessageInput) - if err != nil { - return nil, fmt.Errorf("unable to receive messages: %w", err) + if s.input == nil { + return nil, ErrInputNil } - msgs := make([]messages.Message, len(out.Messages)) - for i, msg := range out.Messages { - msgs[i] = c.createMessage(msg) + output, err := s.client.ReceiveMessage(ctx, s.input) + if err != nil { + return nil, err } - return msgs, nil + return convert(output), nil } -func (c sqsClient) DeleteMessages(messages []messages.Message) error { - _, err := c.svc.DeleteMessageBatch(context.Background(), &awsSqs.DeleteMessageBatchInput{ - Entries: c.prepareMessagesForDeletion(messages), - QueueUrl: c.receiveMessageInput.QueueUrl, +func (s SQSClient) Delete(ctx context.Context, msg Message) error { + if msg.ReceiptHandle == "" { + return ErrReceiptHandleEmpty + } + + _, err := s.client.DeleteMessage(ctx, &sqs.DeleteMessageInput{ + QueueUrl: s.input.QueueUrl, + ReceiptHandle: &msg.ReceiptHandle, }) + if err != nil { + return err + } - return err + return nil } -// prepareMessagesForDeletion takes the processed message batch to transform into a de-duplicated struct for SQS to handle -func (c sqsClient) prepareMessagesForDeletion(messages []messages.Message) []types.DeleteMessageBatchRequestEntry { - deleteEntries := make([]types.DeleteMessageBatchRequestEntry, 0, len(messages)) - processed := map[string]bool{} +func convert(output *sqs.ReceiveMessageOutput) []Message { + messages := make([]Message, 0, len(output.Messages)) - for _, message := range messages { - msgId := message.Msg.(types.Message).MessageId - if _, exists := processed[*msgId]; exists { + for _, msg := range output.Messages { + if msg.MessageId == nil { + // something horrible has happened continue } - deleteEntries = append(deleteEntries, types.DeleteMessageBatchRequestEntry{ - Id: msgId, - ReceiptHandle: message.Msg.(types.Message).ReceiptHandle, - }) - - processed[*msgId] = true - } - - return deleteEntries -} + if msg.Body == nil { + var empty string + msg.Body = &empty + } -func (c sqsClient) createMessage(sqsMessage types.Message) messages.Message { - msg := messages.Message{ - MsgId: *sqsMessage.MessageId, - Msg: sqsMessage, - ReceivedTime: time.Now(), + messages = append(messages, Message{ + ID: *msg.MessageId, + Body: *msg.Body, + ReceiptHandle: *msg.ReceiptHandle, + Attributes: msg.Attributes, + MessageAttributes: convertAttribute(msg.MessageAttributes), + }) } - // Set a context with timeout - msg.Ctx, msg.CancelCtx = context.WithTimeout(context.Background(), c.messageCtxTimeout) - - return msg + return messages } -func NewSqsClient(ctx context.Context, config SqsClientConfiguration) (sqsClient, error) { - messageTimeout := config.MessageCtxTimeout +// convertAttribute turns an AWS type into `any` instead. +func convertAttribute(attributes map[string]types.MessageAttributeValue) map[string]any { + attrs := make(map[string]any) - // Try config.ReceiveMessageInput.VisibilityTimeout first - if messageTimeout == 0 && config.ReceiveMessageInput.VisibilityTimeout != 0 { - messageTimeout = time.Second * time.Duration(config.ReceiveMessageInput.VisibilityTimeout) - } - // Otherwise, infer it from SQS queue's VisibilityTimeout attribute - if messageTimeout == 0 { - var err error - messageTimeout, err = retrieveSqsQueueVisibilityTimeout(ctx, config.Svc, config.ReceiveMessageInput.QueueUrl) - if err != nil { - return sqsClient{}, fmt.Errorf("unable to retrieve visibility timeout: %w", err) + for k, v := range attributes { + if v.DataType == nil { + continue } - } - - return sqsClient{ - svc: config.Svc, - receiveMessageInput: config.ReceiveMessageInput, - messageCtxTimeout: messageTimeout, - }, nil -} -func retrieveSqsQueueVisibilityTimeout(ctx context.Context, svc *awsSqs.Client, queue *string) (time.Duration, error) { - out, err := svc.GetQueueAttributes(ctx, &awsSqs.GetQueueAttributesInput{ - QueueUrl: queue, - AttributeNames: []types.QueueAttributeName{types.QueueAttributeNameVisibilityTimeout}, - }) - if err != nil { - return 0, fmt.Errorf("unable to get attributes: %w", err) - } - - timeout, err := strconv.Atoi(out.Attributes[string(types.QueueAttributeNameVisibilityTimeout)]) - if err != nil { - return 0, fmt.Errorf("unable to parse timeout value: %w", err) + switch *v.DataType { + case "String", "Number": + attrs[k] = *v.StringValue + case "Binary": + attrs[k] = v.BinaryValue + } } - return time.Second * time.Duration(timeout), nil + return attrs } - -// Interface guards -var ( - _ client.Client = (*sqsClient)(nil) -) diff --git a/v2/worker.go b/v2/worker.go deleted file mode 100644 index b0fe2d6..0000000 --- a/v2/worker.go +++ /dev/null @@ -1,106 +0,0 @@ -package formigo - -import ( - "context" - "errors" - "sync" - - "github.com/Pod-Point/go-queue-worker/v2/internal/client" - "github.com/Pod-Point/go-queue-worker/v2/internal/messages" -) - -type Worker struct { - client client.Client - concurrency int - retrievers int - reportFunc func(error) - consumer Consumer - deleterConfig DeleterConfiguration -} - -func (w Worker) Run(ctx context.Context) error { - // Create a controller - ctrl := newController(w.reportFunc) - - // Run retrievers and get the message channel - messageCh := w.runRetrievers(ctx, ctrl) - - // Run consumer and get the deletion channel. - // Note that the context is not given to them because they will only stop once - // all the messages in the pipeline have been consumed. - deleteCh := w.runConsumer(ctrl, messageCh) - - // Run deleter. - // Note that the context is not given to the deleter because it will only stop once - // all the consumed messages in the pipeline have been deleted. - var wg sync.WaitGroup // WaitGroup for the deleter - deleter(&wg, w.client, w.deleterConfig, ctrl, deleteCh) - - // Wait for deleter to exit - wg.Wait() - - // Get the cause of the cancellation - // If the context was cancelled by the controller, there must be a cause containing - // the error, otherwise the context error by default is `context.Canceled`, which - // means that it was a normal stop request (probably SIGTERM). - err := context.Cause(ctx) - if errors.Is(err, context.Canceled) { - return nil - } - - return err -} - -// runRetrievers will run a number of retrievers (Go routines) based on the worker's -// configuration. -// It returns a channel where the messages will be published and, only when all the -// retrievers have stopped, it will close it to broadcast the signal to stop to the -// consumers. -func (w Worker) runRetrievers(ctx context.Context, ctrl *controller) <-chan messages.Message { - messageCh := make(chan messages.Message) - - var wg sync.WaitGroup - wg.Add(w.retrievers) - for i := 0; i < w.retrievers; i++ { - go func() { - defer wg.Done() - retriever(ctx, w.client, ctrl, messageCh) - }() - } - - go func() { - wg.Wait() - close(messageCh) - }() - - return messageCh -} - -// runConsumer runs the worker's consumer. -// It returns a channel where the messages will be published for deletion and, -// only when the consumer has stopped, it will close it to broadcast the -// signal to stop to the deleter. -func (w Worker) runConsumer(ctrl *controller, messageCh <-chan messages.Message) <-chan messages.Message { - deleteCh := make(chan messages.Message) - - go func() { - w.consumer.consume(w.concurrency, ctrl, messageCh, deleteCh) - - close(deleteCh) - }() - - return deleteCh -} - -func NewWorker(config Configuration) Worker { - config = setWorkerConfigValues(config) - - return Worker{ - client: config.Client, - concurrency: config.Concurrency, - retrievers: config.Retrievers, - reportFunc: config.ReportFunc, - consumer: config.Consumer, - deleterConfig: config.DeleterConfig, - } -} diff --git a/v2/worker_test.go b/v2/worker_test.go deleted file mode 100644 index 823b17f..0000000 --- a/v2/worker_test.go +++ /dev/null @@ -1,245 +0,0 @@ -package formigo - -import ( - "context" - "fmt" - "sync" - "testing" - "time" - - "github.com/stretchr/testify/assert" - - "github.com/Pod-Point/go-queue-worker/v2/internal/messages" -) - -type SimpleInMemoryBrokerMessage struct { - messageId string - body string - deleteReqCh chan struct{} - deleteAckCh chan struct{} - timer *time.Timer -} - -type SimpleInMemoryBroker struct { - visibilityTimeout time.Duration - queue chan *SimpleInMemoryBrokerMessage - inFlights chan *SimpleInMemoryBrokerMessage - expired chan *SimpleInMemoryBrokerMessage - - statics struct { - rwMutex sync.RWMutex - enqueuedMessages int - inFlightMessages int - } -} - -func NewSimpleInMemoryBroker(visibilityTimeout time.Duration) *SimpleInMemoryBroker { - return &SimpleInMemoryBroker{ - visibilityTimeout: visibilityTimeout, - queue: make(chan *SimpleInMemoryBrokerMessage, 1000), - inFlights: make(chan *SimpleInMemoryBrokerMessage), - expired: make(chan *SimpleInMemoryBrokerMessage, 1000), - } -} - -func (b *SimpleInMemoryBroker) run(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - case msg := <-b.inFlights: - go func(ctx context.Context) { - select { - case <-ctx.Done(): - return - case <-msg.deleteReqCh: - msg.deleteAckCh <- struct{}{} - case <-msg.timer.C: - b.expired <- msg - } - }(ctx) - } - } -} - -func (b *SimpleInMemoryBroker) AddMessages(msgs []*SimpleInMemoryBrokerMessage) { - for _, msg := range msgs { - b.queue <- msg - b.statics.rwMutex.Lock() - b.statics.enqueuedMessages++ - b.statics.rwMutex.Unlock() - } -} - -func (b *SimpleInMemoryBroker) DeleteMessages(msgs []messages.Message) error { - requestTimer := time.NewTimer(time.Second * 5) - defer requestTimer.Stop() - - for _, msg := range msgs { - brokerMsg := msg.Content().(*SimpleInMemoryBrokerMessage) - - select { - case <-requestTimer.C: - return fmt.Errorf("failed to delete message %s: request timeout", brokerMsg.messageId) - case brokerMsg.deleteReqCh <- struct{}{}: - } - - if !brokerMsg.timer.Stop() { - return fmt.Errorf("failed to delete message %s: visibility timeout exipired", brokerMsg.messageId) - } - - <-brokerMsg.deleteAckCh - - b.statics.rwMutex.Lock() - b.statics.inFlightMessages-- - b.statics.rwMutex.Unlock() - } - - return nil -} - -func (b *SimpleInMemoryBroker) ReceiveMessages(ctx context.Context) ([]messages.Message, error) { - var polledMessage *SimpleInMemoryBrokerMessage - select { - case polledMessage = <-b.expired: - default: - timer := time.NewTimer(time.Millisecond * 500) - defer timer.Stop() - - select { - case <-ctx.Done(): - return nil, ctx.Err() - case <-timer.C: - return nil, nil - case polledMessage = <-b.expired: - case polledMessage = <-b.queue: - } - } - - polledMessage.timer = time.NewTimer(b.visibilityTimeout) - polledMessage.deleteReqCh = make(chan struct{}) - polledMessage.deleteAckCh = make(chan struct{}) - - time.After(time.Millisecond * 5) - - msg := messages.Message{ - MsgId: polledMessage.messageId, - Msg: polledMessage, - ReceivedTime: time.Now(), - } - - // Set a context with timeout - msg.Ctx, msg.CancelCtx = context.WithTimeout(context.Background(), b.visibilityTimeout) - - // Move the message to inflight - b.inFlights <- polledMessage - b.statics.rwMutex.Lock() - b.statics.enqueuedMessages-- - b.statics.inFlightMessages++ - b.statics.rwMutex.Unlock() - - return []messages.Message{msg}, nil -} - -func (b *SimpleInMemoryBroker) EnqueuedMessages() int { - b.statics.rwMutex.RLock() - defer b.statics.rwMutex.RUnlock() - return b.statics.enqueuedMessages -} - -func (b *SimpleInMemoryBroker) InFlightMessages() int { - b.statics.rwMutex.RLock() - defer b.statics.rwMutex.RUnlock() - return b.statics.inFlightMessages -} - -func TestWorker(t *testing.T) { - inMemoryBroker := NewSimpleInMemoryBroker(time.Second * 10) - go inMemoryBroker.run(context.Background()) - - t.Run("can receive a message", func(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - msgs := []*SimpleInMemoryBrokerMessage{ - { - messageId: "1", - body: "Hello, world!", - }, - } - - inMemoryBroker.AddMessages(msgs) - - wkr := NewWorker(Configuration{ - Client: inMemoryBroker, - Concurrency: 1, - Retrievers: 1, - ReportFunc: func(err error) { - t.Fatalf("unexpected error: %v", err) - }, - Consumer: NewMessageConsumer(MessageConsumerConfiguration{ - Handler: func(ctx context.Context, msg Message) error { - defer cancel() - - assert.Equal(t, "Hello, world!", msg.Content().(*SimpleInMemoryBrokerMessage).body) - - return nil - }, - }), - }) - - assert.NoError(t, wkr.Run(ctx)) - assert.Equal(t, 0, inMemoryBroker.EnqueuedMessages()) - assert.Equal(t, 0, inMemoryBroker.InFlightMessages()) - }) - - t.Run("can receive a batch of messages", func(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - msgs := []*SimpleInMemoryBrokerMessage{ - { - messageId: "1", - body: "Hello, world 1!", - }, - { - messageId: "2", - body: "Hello, world 2!", - }, - { - messageId: "3", - body: "Hello, world 3!", - }, - } - - inMemoryBroker.AddMessages(msgs) - - wkr := NewWorker(Configuration{ - Client: inMemoryBroker, - Concurrency: 1, - Retrievers: 1, - ReportFunc: func(err error) { - t.Fatalf("unexpected error: %v", err) - }, - Consumer: NewBatchConsumer(BatchConsumerConfiguration{ - BufferConfig: BatchConsumerBufferConfiguration{ - Size: 3, - Timeout: time.Second, - }, - Handler: func(ctx context.Context, msgs []Message) (BatchResponse, error) { - defer cancel() - - if len(msgs) < 3 { - t.Fatalf("expected 3 messages, got %d", len(msgs)) - } - - assert.Equal(t, "Hello, world 1!", msgs[0].Content().(*SimpleInMemoryBrokerMessage).body) - assert.Equal(t, "Hello, world 2!", msgs[1].Content().(*SimpleInMemoryBrokerMessage).body) - assert.Equal(t, "Hello, world 3!", msgs[2].Content().(*SimpleInMemoryBrokerMessage).body) - - return BatchResponse{}, nil - }, - }), - }) - - assert.NoError(t, wkr.Run(ctx)) - assert.Equal(t, 0, inMemoryBroker.EnqueuedMessages()) - assert.Equal(t, 0, inMemoryBroker.InFlightMessages()) - }) -} From 5722437a52a1527980b46ba50ba8a9ecbcb01069 Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Wed, 30 Sep 2026 12:58:33 +0100 Subject: [PATCH 04/11] feat(formigo/v2): rework the entire queue worker arch to use pond --- v2/README.md | 58 ++++++++++++++++- v2/cmd/echo/main.go | 82 +++++++++++++++++++++++ v2/cmd/producer/main.go | 59 +++++++++++++++++ v2/go.mod | 20 ++++-- v2/go.sum | 36 +++++++--- v2/manager.go | 141 ++++++++++++++++++++++++++++++++++------ v2/manager_test.go | 48 ++++++++++++++ v2/options.go | 77 ++++++++++++++++++++++ v2/sqs.go | 7 ++ 9 files changed, 493 insertions(+), 35 deletions(-) create mode 100644 v2/cmd/echo/main.go create mode 100644 v2/cmd/producer/main.go create mode 100644 v2/manager_test.go create mode 100644 v2/options.go diff --git a/v2/README.md b/v2/README.md index 3169cad..99516f3 100644 --- a/v2/README.md +++ b/v2/README.md @@ -9,4 +9,60 @@ Basic features of Pond: - automatic scaling to available resources/limits based on incoming queue pressure. - fire & forget queue submission - fire & **wait for a response** submission to a queue (e.g. for dependent jobs, or deferred behaviour). -- clean, graceful exit when shutting down. \ No newline at end of file +- clean, graceful exit when shutting down. + +## Usage + +In `main()`, you'll generally have the following: + +```go +package main + +import ( + "context" + "log" + "time" + + "libs/go/application" + + formigo "github.com/Pod-Point/go-queue-worker/v2" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/sqs" +) + +func main() { + // app here has a context that uses signal.NotifyContext to catch SIGTERM/SIGKILL/SIGHUP etc. + app := application.New().HandlesSigTerm() + + app.Run(func(ctx context.Context) error { + cfg, err := config.LoadDefaultConfig(ctx) + if err != nil { + return err + } + + client := formigo.NewSQSClient(sqs.NewFromConfig(cfg), &sqs.ReceiveMessageInput{ + QueueUrl: aws.String("https://sqs.eu-west-1.amazonaws.com/123456789012/my-queue"), + MaxNumberOfMessages: 10, + VisibilityTimeout: 30, + WaitTimeSeconds: 20, + }) + + manager := formigo.NewManager(client, + formigo.WithDeadline(time.Second * 45), + formigo.WithFetchConcurrency(2), + formigo.WithFetchDelay(time.Second * 1), + formigo.WithWorkerConcurrency(20), + formigo.WithReporter(func(err error) { + log.Print(err) // send to sentry, structured logging, etc + }), + formigo.WithConsumer(func(ctx context.Context, msg formigo.Message) error { + return nil + }), + ) + + return manager.Run(ctx) + }) +} + +``` \ No newline at end of file diff --git a/v2/cmd/echo/main.go b/v2/cmd/echo/main.go new file mode 100644 index 0000000..2a66bd8 --- /dev/null +++ b/v2/cmd/echo/main.go @@ -0,0 +1,82 @@ +// Package main is a test consumer for a localstack queue. +// This is not meant to be a production config, though it is set up exactly as you would use it there. +package main + +import ( + "context" + "errors" + "fmt" + "log" + "log/slog" + "os" + "os/signal" + "syscall" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/sqs" + + formigo "github.com/Pod-Point/go-queue-worker/v2" +) + +type Message struct { + Level string `json:"level"` + Message string `json:"message"` +} + +func main() { + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGILL, syscall.SIGHUP) + defer cancel() + + cfg, err := config.LoadDefaultConfig(ctx) + if err != nil { + log.Printf("unable to load config: %v\n", err) + os.Exit(2) + } + + client := formigo.NewSQSClient(sqs.NewFromConfig(cfg), &sqs.ReceiveMessageInput{ + QueueUrl: aws.String("http://sqs.eu-west-1.localhost.localstack.cloud:4566/000000000000/formigo-echo-testing"), + MaxNumberOfMessages: 5, + VisibilityTimeout: 6, + WaitTimeSeconds: 10, + }) + + manager := formigo.NewManager(client, + formigo.WithDeadline(time.Second*4), + formigo.WithFetchConcurrency(2), + formigo.WithFetchDelay(time.Second*1), + formigo.WithWorkerConcurrency(3), + formigo.WithReporter(func(err error) { + slog.Error(err.Error(), slog.String("type", fmt.Sprintf("%T", err))) + }), + formigo.WithConsumer(func(ctx context.Context, msg formigo.Message) error { + var body Message + if err := msg.Decode(&body); err != nil { + return err + } + + switch body.Level { + case "error": + return errors.New(body.Message) + case "info": + slog.InfoContext(ctx, body.Message) + case "warn", "warning": + slog.WarnContext(ctx, body.Message) + case "debug": + slog.DebugContext(ctx, body.Message) + } + + return nil + }), + formigo.WithLogger(slog.New( + slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{ + Level: slog.LevelInfo, + }), + )), + ) + + if err := manager.Run(ctx); err != nil { + log.Fatal(err) + } +} diff --git a/v2/cmd/producer/main.go b/v2/cmd/producer/main.go new file mode 100644 index 0000000..c00d66d --- /dev/null +++ b/v2/cmd/producer/main.go @@ -0,0 +1,59 @@ +// Package main is a test producer for a localstack queue. +// This is not meant to be a production config, though it is set up exactly as you would use it there. +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "os" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/sqs" +) + +var usage = "USAGE: ./echo [level] \"[message]\"" + +func main() { + ctx := context.Background() + + cfg, err := config.LoadDefaultConfig(ctx) + if err != nil { + log.Printf("unable to load config: %v\n", err) + os.Exit(2) + } + + if len(os.Args[1:]) != 2 { + fmt.Printf(usage) + } + + level := os.Args[1] + body := os.Args[2] + + b, err := json.Marshal(struct { + Level string `json:"level"` + Body string `json:"message"` + }{ + Level: level, + Body: body, + }) + if err != nil { + log.Fatal(err) + } + + client := sqs.NewFromConfig(cfg) + + // duplicate for testing + for range 10 { + output, err := client.SendMessage(ctx, &sqs.SendMessageInput{ + MessageBody: aws.String(string(b)), + QueueUrl: aws.String("http://sqs.eu-west-1.localhost.localstack.cloud:4566/000000000000/formigo-echo-testing"), + }) + if err != nil { + log.Fatal(err) + } + fmt.Printf("message id: %s\n", *output.MessageId) + } +} diff --git a/v2/go.mod b/v2/go.mod index 2d49cc2..f35f577 100644 --- a/v2/go.mod +++ b/v2/go.mod @@ -3,16 +3,26 @@ module github.com/Pod-Point/go-queue-worker/v2 go 1.26 require ( + github.com/alitto/pond/v2 v2.7.1 + github.com/aws/aws-sdk-go-v2 v1.47.1 + github.com/aws/aws-sdk-go-v2/config v1.33.6 github.com/aws/aws-sdk-go-v2/service/sqs v1.34.5 github.com/stretchr/testify v1.9.0 ) require ( - github.com/alitto/pond/v2 v2.7.1 // indirect - github.com/aws/aws-sdk-go-v2 v1.30.4 // indirect - github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.16 // indirect - github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.16 // indirect - github.com/aws/smithy-go v1.20.4 // indirect + github.com/aws/aws-sdk-go-v2/credentials v1.20.6 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.1 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.4 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.4 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.4 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.4 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.10.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.38.1 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.43.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.51.1 // indirect + github.com/aws/smithy-go v1.28.1 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/v2/go.sum b/v2/go.sum index bc591a3..a940922 100644 --- a/v2/go.sum +++ b/v2/go.sum @@ -1,15 +1,35 @@ github.com/alitto/pond/v2 v2.7.1 h1:QxMbcfjcVTa0pyxX5Ib1226mM8u8D7gKUVkCUU4DYIw= github.com/alitto/pond/v2 v2.7.1/go.mod h1:xkjYEgQ05RSpWdfSd1nM3OVv7TBhLdy7rMp3+2Nq+yE= -github.com/aws/aws-sdk-go-v2 v1.30.4 h1:frhcagrVNrzmT95RJImMHgabt99vkXGslubDaDagTk8= -github.com/aws/aws-sdk-go-v2 v1.30.4/go.mod h1:CT+ZPWXbYrci8chcARI3OmI/qgd+f6WtuLOoaIA8PR0= -github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.16 h1:TNyt/+X43KJ9IJJMjKfa3bNTiZbUP7DeCxfbTROESwY= -github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.16/go.mod h1:2DwJF39FlNAUiX5pAc0UNeiz16lK2t7IaFcm0LFHEgc= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.16 h1:jYfy8UPmd+6kJW5YhY0L1/KftReOGxI/4NtVSTh9O/I= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.16/go.mod h1:7ZfEPZxkW42Afq4uQB8H2E2e6ebh6mXTueEpYzjCzcs= +github.com/aws/aws-sdk-go-v2 v1.47.1 h1:uOIZnp4PK3ZhKI0dNrJrhTEsLxbpXHTAJlwoS1pvAtw= +github.com/aws/aws-sdk-go-v2 v1.47.1/go.mod h1:bttEH6JqnUL8LepvDVfdrds/fZ5bCIxzpe3abyUrhDU= +github.com/aws/aws-sdk-go-v2/config v1.33.6 h1:MBjkSTLczek/UgiK+EYPIoRTqE7gP8vtW3OFbFo7Nug= +github.com/aws/aws-sdk-go-v2/config v1.33.6/go.mod h1:grRAFzdAZJrwcbasJRg2MPvIrVjtlfXllHssN6+E1JE= +github.com/aws/aws-sdk-go-v2/credentials v1.20.6 h1:NpAFXCU7NzXNkdGK3zQTtsRJ+3v9tZQV0xcdRw8uBdw= +github.com/aws/aws-sdk-go-v2/credentials v1.20.6/go.mod h1:mcZCoiPnyMvP8VMNbygNX5lLqSlkYJIMPODylQMurOk= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.1 h1:8gALAAmacnIXh+z6VkdDanv4/IkG5APdg4DZLDTmLog= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.1/go.mod h1:Z7IJhJU+poOdJjUR2wpyY21ossQ1XS/R3Lk9Msq5kM4= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.4 h1:CLq4+8UHCI+ZZYl/EuJxXovaIVN2xeeT8JV+dsApQ5E= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.4/go.mod h1:Wv4q5sAM04xAMkoOedxLx2inVf6K5FdxYp+A61L+q/0= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.4 h1:dD4MR81I7YkpEBRk6UP9rocC2QnT3qVuXwzlYTtfGEs= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.4/go.mod h1:EcXV1kAFd5XwSkDHlj94gnF3q5CkJyYiIJfH8N0VmrE= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.4 h1:7Wo47d/xn/7KttCSBd8EGYeZ7ULRFRkUHr6vkZPBzVQ= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.4/go.mod h1:tDB2IVC1xC3vX8o+6uRlzhTxP3g1b77CZXFX/oD2FnQ= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 h1:bAdDl/HkGCcGPoe25ToSHEw23VIxt6CT5fLcg111BKg= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19/go.mod h1:KaUzbLxv4CeSxh6ZCl9B4m7CuFenS8kUEaDs+f/DQr4= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.4 h1:29SvnfGhXjTl8ONxFwbj2rs6lbhiFXD2CgFQmbT/bXY= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.4/go.mod h1:wm04I5DMuNVvZHFe/dHnUxincvNbbK7AiNBbYsQivek= +github.com/aws/aws-sdk-go-v2/service/signin v1.10.1 h1:DzCCWLzcIRQ77F3DEUljud7bEjTgFOIKXP52NmVRyhU= +github.com/aws/aws-sdk-go-v2/service/signin v1.10.1/go.mod h1:xpo/geVldu8payT375WekctUzopG/hBU7miiqItMUlw= github.com/aws/aws-sdk-go-v2/service/sqs v1.34.5 h1:HYyVDOC2/PIg+3oBX1q0wtDU5kONki6lrgIG0afrBkY= github.com/aws/aws-sdk-go-v2/service/sqs v1.34.5/go.mod h1:7idt3XszF6sE9WPS1GqZRiDJOxw4oPtlRBXodWnCGjU= -github.com/aws/smithy-go v1.20.4 h1:2HK1zBdPgRbjFOHlfeQZfpC4r72MOb9bZkiFwggKO+4= -github.com/aws/smithy-go v1.20.4/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg= +github.com/aws/aws-sdk-go-v2/service/sso v1.38.1 h1:Umtl/0YZhng4xndfW3lKJrYYP7NLEjI6bGXVomwLcs0= +github.com/aws/aws-sdk-go-v2/service/sso v1.38.1/go.mod h1:rRD/dnm7q0HYE/I5TMaPgkWyyUGLcwuxHLABsLnQ3e0= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.43.1 h1:orIWdNiLgzrhu/11RcPPKO/SBzUUymbUQuZbSPImghg= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.43.1/go.mod h1:skwM/xsbR/1ReUTesv9BhpJp1VjajR7DWQnuVLwiXsQ= +github.com/aws/aws-sdk-go-v2/service/sts v1.51.1 h1:0HOqZXRvMytH6bFHVIc0oJX07sZjfhz0zXtjs6gdE8s= +github.com/aws/aws-sdk-go-v2/service/sts v1.51.1/go.mod h1:26zA0GhDrLo+yiLI2yXWxqB1PdsShfLikoI7GOEgugM= +github.com/aws/smithy-go v1.28.1 h1:R/nXH00c8qcfCzQVELtRw+eLQWtzv+VAIEFJ1/xxXlQ= +github.com/aws/smithy-go v1.28.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= diff --git a/v2/manager.go b/v2/manager.go index ca1deab..8348257 100644 --- a/v2/manager.go +++ b/v2/manager.go @@ -2,11 +2,17 @@ package formigo import ( "context" + "errors" + "log/slog" "time" "github.com/alitto/pond/v2" ) +var ( + ErrConsumerMissing = errors.New("formigo/v2: consumer is not set. Use .WithConsumer()") +) + // Fetcher condenses RetrieveMessage into the smallest possible interface. type Fetcher interface { Fetch(context.Context) ([]Message, error) @@ -17,6 +23,7 @@ type Deleter interface { Delete(context.Context, Message) error } +// Client is a thin wrapper around an SQS client that can fetch and delete messages. type Client interface { Fetcher Deleter @@ -24,66 +31,112 @@ type Client interface { // Manager sets up a set of workers and processes an queue via them. type Manager struct { - client Client - pool pond.Pool - deadline time.Duration - fetchDelay time.Duration - - deleter func(context.Context, Message) error - consumer func(context.Context, Message) error - reporter func(error) + client Client + pool pond.Pool + deadline time.Duration + fetchDelay time.Duration + fetchConcurrency int + workerConcurrency int + consumer func(context.Context, Message) error + reporter func(error) + logger *slog.Logger } // Submit will push the message onto a consumer via the worker pool. -func (m *Manager) Submit(msg Message) error { +func (m *Manager) Submit(msg Message) (err error) { m.pool.SubmitErr(func() error { ctx := context.Background() // make sure this consumer deletes the message from the queue when deferring. // this one uses the background context with no deadline, and reports its own errors. - defer m.Delete(ctx, msg) + defer m.Delete(ctx, msg, err) + + m.Log(ctx, slog.LevelDebug, "processing message", + slog.String("messageID", msg.ID), + slog.String("receiptHandle", msg.ReceiptHandle), + ) if m.deadline > 0 { + m.Log(ctx, slog.LevelDebug, "applying deadline", + slog.String("messageID", msg.ID), + slog.Duration("deadline", m.deadline), + ) + var cancel func() ctx, cancel = context.WithDeadline(ctx, time.Now().Add(m.deadline)) defer cancel() } - if err := m.consumer(ctx, msg); err != nil { - m.reporter(err) - return err + err = m.consumer(ctx, msg) + + if err != nil { + m.Log(ctx, slog.LevelInfo, "message: errored", + slog.String("messageID", msg.ID), + slog.String("error", err.Error()), + ) + } else { + m.Log(ctx, slog.LevelInfo, "message: processed", + slog.String("messageID", msg.ID), + ) } - return nil + return err }) return nil } -// Delete will call the Client's Delete method, skipping any batch deletes. -func (m *Manager) Delete(ctx context.Context, msg Message) { +// Delete will call the Client's Delete method and remove a message from the queue. +// This may succeed with no error, but still not remove the message (e.g. it was not the most recent fetch of this message). +func (m *Manager) Delete(ctx context.Context, msg Message, err error) { + m.Log(ctx, slog.LevelDebug, "deleting message", + slog.String("messageID", msg.ID), + slog.String("receiptHandle", msg.ReceiptHandle), + ) + + if err != nil { + m.Report(err) + } + + // if err is a failure, we can choose if we retry here. + // TODO: retry strategy + if err := m.client.Delete(ctx, msg); err != nil { - m.reporter(err) + m.Log(ctx, slog.LevelError, "failed to delete message", + slog.String("messageID", msg.ID), + slog.String("error", err.Error()), + ) + m.Report(err) + } else { + m.Log(ctx, slog.LevelInfo, "message deleted", + slog.String("messageID", msg.ID), + ) } } +// Fetch will run a continuous loop of querying SQS for messages. func (m *Manager) Fetch(ctx context.Context) error { for { select { case <-ctx.Done(): default: + m.Log(ctx, slog.LevelDebug, "fetching messages") messages, err := m.client.Fetch(ctx) if err != nil { m.Report(err) // report errors with the initial fetch. continue } + m.Log(ctx, slog.LevelDebug, "fetched messages", slog.Int("count", len(messages))) + for _, msg := range messages { if err := m.Submit(msg); err != nil { m.Report(err) } } + m.Log(ctx, slog.LevelDebug, "waiting before polling", slog.Duration("delay", m.fetchDelay)) + // allow configuring a delay after fetching messages // this may help prevent excessive pressure on AWS. time.Sleep(m.fetchDelay) @@ -99,14 +152,22 @@ func (m *Manager) Report(err error) { } // Run will start up a worker pool and start processing a queue. -// It must exit via the context, e.g. via signal.NotifyContext. +// It must exit via the context, e.g. via signal.NotifyContext or cancel(). // Using context.Background will never safely exit. +// +// This will error if a consumer is not set. func (m *Manager) Run(ctx context.Context) error { - m.pool = pond.NewPool(20, pond.WithContext(ctx)) + if m.consumer == nil { + return ErrConsumerMissing + } + + m.pool = pond.NewPool(m.workerConcurrency, pond.WithContext(ctx)) // set up the fetchers to feed the workers and start them - fetcher := pond.NewPool(2, pond.WithContext(ctx)) - for range 2 { + fetcher := pond.NewPool(m.fetchConcurrency, pond.WithContext(ctx)) + for i := range m.fetchConcurrency { + m.Log(ctx, slog.LevelInfo, "starting fetcher", slog.Int("id", i)) + fetcher.SubmitErr(func() error { return m.Fetch(ctx) }) @@ -114,11 +175,49 @@ func (m *Manager) Run(ctx context.Context) error { select { case <-ctx.Done(): + m.Log(ctx, slog.LevelInfo, "stopping fetchers") + // stop the fetcher first to be sure it has drained fetcher.StopAndWait() + m.Log(ctx, slog.LevelInfo, "stopping workers") + // stop and drain the worker pool m.pool.StopAndWait() return ctx.Err() } } + +func (m *Manager) Log(ctx context.Context, level slog.Level, msg string, args ...any) { + if m.logger != nil { + m.logger.Log(ctx, level, msg, args...) + } +} + +// NewManager will create a new *Manager with the given client and options. +func NewManager(client Client, opts ...Option) *Manager { + manager := NewDefaultManager(client) + + for _, opt := range opts { + opt.Apply(manager) + } + + manager.Log(context.Background(), slog.LevelInfo, "applied options", + slog.Int("workerConcurrency", manager.workerConcurrency), + slog.Int("fetchConcurrency", manager.fetchConcurrency), + slog.Duration("fetchDelay", manager.fetchDelay), + slog.Duration("deadline", manager.deadline), + ) + + return manager +} + +func NewDefaultManager(client Client) *Manager { + return &Manager{ + client: client, + fetchDelay: time.Millisecond * 500, + fetchConcurrency: 2, + workerConcurrency: 20, + deadline: time.Second * 30, + } +} diff --git a/v2/manager_test.go b/v2/manager_test.go new file mode 100644 index 0000000..aafbf29 --- /dev/null +++ b/v2/manager_test.go @@ -0,0 +1,48 @@ +package formigo + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +type TestClient struct { + // +} + +func (t TestClient) Fetch(ctx context.Context) ([]Message, error) { + return nil, nil +} + +func (t TestClient) Delete(ctx context.Context, message Message) error { + return nil +} + +var _ Client = (*TestClient)(nil) + +// testContext is a safety measure for testing Run in case there is never an exit condition. +func testContext() (context.Context, context.CancelFunc) { + return context.WithDeadline(context.Background(), time.Now().Add(time.Second*4)) +} + +func TestNewDefaultManager_VerifiesDefaults(t *testing.T) { + manager := NewDefaultManager(TestClient{}) + + assert.Nil(t, manager.consumer) + assert.Nil(t, manager.reporter) + + assert.Equal(t, 2, manager.fetchConcurrency) + assert.Equal(t, 20, manager.workerConcurrency) + assert.Equal(t, time.Millisecond*500, manager.fetchDelay) + assert.Equal(t, time.Second*30, manager.deadline) +} + +func TestManager_Run_FailsWithoutConsumer(t *testing.T) { + manager := NewDefaultManager(TestClient{}) + ctx, cancel := testContext() + defer cancel() + + assert.ErrorIs(t, manager.Run(ctx), ErrConsumerMissing) +} diff --git a/v2/options.go b/v2/options.go new file mode 100644 index 0000000..7f890c1 --- /dev/null +++ b/v2/options.go @@ -0,0 +1,77 @@ +package formigo + +import ( + "context" + "log/slog" + "time" +) + +type Option interface { + Apply(*Manager) +} + +type OptionFunc func(*Manager) + +func (f OptionFunc) Apply(manager *Manager) { + f(manager) +} + +// WithDeadline sets the context deadline used for consumer processes when they are given a message. +// Ensure this is lower than VisibilityTimeout in SQS, or you may process the same message twice. +// +// Default: 30s +func WithDeadline(deadline time.Duration) OptionFunc { + return func(manager *Manager) { + manager.deadline = deadline + } +} + +// WithWorkerConcurrency will set the number of workers used to run a consumer process. +// Set this high enough so that you can process a reasonably high number of messages, +// but not so high that you throttle/cpu-limit the other workers. +// +// Default: 20 +func WithWorkerConcurrency(concurrency int) OptionFunc { + return func(manager *Manager) { + manager.workerConcurrency = concurrency + } +} + +// WithFetchConcurrency will set the number of workers that will be continuously fetching messages from the given queue. +// Do not set this significantly higher than needed (3-4 is sufficient) or you may grow the queue backlog. +// +// Default: 2 +func WithFetchConcurrency(concurrency int) OptionFunc { + return func(manager *Manager) { + manager.fetchConcurrency = concurrency + } +} + +// WithFetchDelay will set the amount of time to sleep between fetching new messages. +// Default: 500ms +func WithFetchDelay(delay time.Duration) OptionFunc { + return func(manager *Manager) { + manager.fetchDelay = delay + } +} + +// WithReporter will set the error reporting function (e.g. log +func WithReporter(f func(error)) OptionFunc { + return func(manager *Manager) { + manager.reporter = f + } +} + +// WithConsumer sets the consumer to use when processing messages. +// Required; panics if not set since this is a dev issue. +func WithConsumer(f func(context.Context, Message) error) OptionFunc { + return func(manager *Manager) { + manager.consumer = f + } +} + +func WithLogger(logger *slog.Logger) OptionFunc { + return func(manager *Manager) { + manager.logger = logger + } +} diff --git a/v2/sqs.go b/v2/sqs.go index 71211ae..1678331 100644 --- a/v2/sqs.go +++ b/v2/sqs.go @@ -52,6 +52,13 @@ func (s SQSClient) Delete(ctx context.Context, msg Message) error { return nil } +func NewSQSClient(client *sqs.Client, input *sqs.ReceiveMessageInput) *SQSClient { + return &SQSClient{ + client: client, + input: input, + } +} + func convert(output *sqs.ReceiveMessageOutput) []Message { messages := make([]Message, 0, len(output.Messages)) From ee5def73cf1f43f2777382b9d3ee94213b5dc70e Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Wed, 30 Sep 2026 13:47:17 +0100 Subject: [PATCH 05/11] feat(formigo/v2): update the readme for v2 --- v2/README.md | 94 +++++++++++++++++++++++++--------------------------- 1 file changed, 45 insertions(+), 49 deletions(-) diff --git a/v2/README.md b/v2/README.md index 99516f3..21621df 100644 --- a/v2/README.md +++ b/v2/README.md @@ -13,56 +13,52 @@ Basic features of Pond: ## Usage -In `main()`, you'll generally have the following: +See `cmd/echo/main.go` for a complete example of how to use v2. + +### Consumers + +Your consumer should adhere to the following: ```go -package main - -import ( - "context" - "log" - "time" - - "libs/go/application" - - formigo "github.com/Pod-Point/go-queue-worker/v2" - "github.com/aws/aws-sdk-go-v2/aws" - "github.com/aws/aws-sdk-go-v2/config" - "github.com/aws/aws-sdk-go-v2/service/sqs" -) - -func main() { - // app here has a context that uses signal.NotifyContext to catch SIGTERM/SIGKILL/SIGHUP etc. - app := application.New().HandlesSigTerm() - - app.Run(func(ctx context.Context) error { - cfg, err := config.LoadDefaultConfig(ctx) - if err != nil { - return err - } - - client := formigo.NewSQSClient(sqs.NewFromConfig(cfg), &sqs.ReceiveMessageInput{ - QueueUrl: aws.String("https://sqs.eu-west-1.amazonaws.com/123456789012/my-queue"), - MaxNumberOfMessages: 10, - VisibilityTimeout: 30, - WaitTimeSeconds: 20, - }) - - manager := formigo.NewManager(client, - formigo.WithDeadline(time.Second * 45), - formigo.WithFetchConcurrency(2), - formigo.WithFetchDelay(time.Second * 1), - formigo.WithWorkerConcurrency(20), - formigo.WithReporter(func(err error) { - log.Print(err) // send to sentry, structured logging, etc - }), - formigo.WithConsumer(func(ctx context.Context, msg formigo.Message) error { - return nil - }), - ) - - return manager.Run(ctx) - }) +import formigo "github.com/Pod-Point/go-queue-worker/v2" + +func (context.Context, formigo.Message) error +``` + +A `Decode` method is provided on formigo.Message to help ensure proper decoding with JSON SQS messages: + +```go +type Message struct { + Foo string `json:"foo"` +} + +func (ctx context.Context, msg formigo.Message) error { + var body Message + if err := msg.Decode(&body); err != nil { + return err + } + // pass to your internal consumer, etc. } +``` + +### Error Reporting + +If you need errors to be sent to Sentry, structured logging, etc, feel free to use `WithReporter` when setting up a manager. + +```go +formigo.WithReporter(func(err error) { + sentry.CaptureException(err) +}) +``` + +### Other Options + +All other options should be fairly self-explanatory, and have godocs and defaults. + +## Shutting Down + +Ensure the context passed to `manager.Run(ctx)` is cancelable, preferrably via `signal.NotifyContext`. + +If it isn't, the program will have no exit condition until fully terminated by the operating system. -``` \ No newline at end of file +See the example consumer to see how this is done. \ No newline at end of file From e19fcd110b7814498feffe413dee56d19a65b26c Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Wed, 30 Sep 2026 13:56:28 +0100 Subject: [PATCH 06/11] feat(formigo/v2): update godocs for v2 --- v2/manager.go | 4 ++++ v2/sqs.go | 5 +++++ 2 files changed, 9 insertions(+) diff --git a/v2/manager.go b/v2/manager.go index 8348257..99eb3ff 100644 --- a/v2/manager.go +++ b/v2/manager.go @@ -1,3 +1,5 @@ +// Package formigo exposes an extremely simple way to run SQS messages inside a worker pool with capped concurrency. +// It also handles graceful shutdown, and ensures messages in-flight are completed on context cancellation. package formigo import ( @@ -188,6 +190,7 @@ func (m *Manager) Run(ctx context.Context) error { } } +// Log will send a log record to a configured slog.Logger, if set. func (m *Manager) Log(ctx context.Context, level slog.Level, msg string, args ...any) { if m.logger != nil { m.logger.Log(ctx, level, msg, args...) @@ -212,6 +215,7 @@ func NewManager(client Client, opts ...Option) *Manager { return manager } +// NewDefaultManager returns a Manager with all defaults set in the options. func NewDefaultManager(client Client) *Manager { return &Manager{ client: client, diff --git a/v2/sqs.go b/v2/sqs.go index 1678331..39c417c 100644 --- a/v2/sqs.go +++ b/v2/sqs.go @@ -14,11 +14,13 @@ var ( ErrReceiptHandleEmpty = errors.New("formigo/v2: message ReceiptHandle is empty when deleting") ) +// SQSClient is an implementation of Client, and is a thin wrapper around AWS SQS. type SQSClient struct { client *sqs.Client input *sqs.ReceiveMessageInput } +// Fetch will attempt to fetch SQS messages from the configured queue, and convert them to Message. func (s SQSClient) Fetch(ctx context.Context) ([]Message, error) { if s.client == nil { return nil, ErrClientNil @@ -36,6 +38,7 @@ func (s SQSClient) Fetch(ctx context.Context) ([]Message, error) { return convert(output), nil } +// Delete takes a message and deletes it from the queue. func (s SQSClient) Delete(ctx context.Context, msg Message) error { if msg.ReceiptHandle == "" { return ErrReceiptHandleEmpty @@ -52,6 +55,7 @@ func (s SQSClient) Delete(ctx context.Context, msg Message) error { return nil } +// NewSQSClient makes a new SQSClient from an AWS SQS client and the parameters used for ReceiveMessage. func NewSQSClient(client *sqs.Client, input *sqs.ReceiveMessageInput) *SQSClient { return &SQSClient{ client: client, @@ -59,6 +63,7 @@ func NewSQSClient(client *sqs.Client, input *sqs.ReceiveMessageInput) *SQSClient } } +// convert will take sqs output and convert it into a slice of Message. func convert(output *sqs.ReceiveMessageOutput) []Message { messages := make([]Message, 0, len(output.Messages)) From 4da2b4d02f9479da51e51eb17d09fd8105c319f9 Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Wed, 30 Sep 2026 13:58:40 +0100 Subject: [PATCH 07/11] feat(formigo/v2): update producer/main.go's usage --- v2/cmd/producer/main.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/v2/cmd/producer/main.go b/v2/cmd/producer/main.go index c00d66d..0721823 100644 --- a/v2/cmd/producer/main.go +++ b/v2/cmd/producer/main.go @@ -14,8 +14,6 @@ import ( "github.com/aws/aws-sdk-go-v2/service/sqs" ) -var usage = "USAGE: ./echo [level] \"[message]\"" - func main() { ctx := context.Background() @@ -26,7 +24,7 @@ func main() { } if len(os.Args[1:]) != 2 { - fmt.Printf(usage) + fmt.Println("USAGE: ./echo [level] \"[message]\"") } level := os.Args[1] From c23f83268de55387a441709269fd81d6131f498c Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Thu, 1 Oct 2026 09:13:08 +0100 Subject: [PATCH 08/11] feat(formigo/v2): properly exit fetcher's pool --- v2/manager.go | 1 + 1 file changed, 1 insertion(+) diff --git a/v2/manager.go b/v2/manager.go index 99eb3ff..0b0cd25 100644 --- a/v2/manager.go +++ b/v2/manager.go @@ -121,6 +121,7 @@ func (m *Manager) Fetch(ctx context.Context) error { for { select { case <-ctx.Done(): + return ctx.Err() default: m.Log(ctx, slog.LevelDebug, "fetching messages") messages, err := m.client.Fetch(ctx) From ca911b0eb9b1b9c9e44a7a56526fba01aeabdb38 Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Thu, 1 Oct 2026 10:03:57 +0100 Subject: [PATCH 09/11] feat(formigo/v2): wrap the delete defer to make sure we execute it --- v2/manager.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/v2/manager.go b/v2/manager.go index 0b0cd25..f46b6e0 100644 --- a/v2/manager.go +++ b/v2/manager.go @@ -51,7 +51,10 @@ func (m *Manager) Submit(msg Message) (err error) { // make sure this consumer deletes the message from the queue when deferring. // this one uses the background context with no deadline, and reports its own errors. - defer m.Delete(ctx, msg, err) + defer func() { + // wrapped in a func() to defer scope capture + m.Delete(ctx, msg, err) + }() m.Log(ctx, slog.LevelDebug, "processing message", slog.String("messageID", msg.ID), From 91cbb01d8f1837b850b594288e0d2335804ce029 Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Thu, 1 Oct 2026 12:10:48 +0100 Subject: [PATCH 10/11] feat(formigo/v2): do not delete context.Canceled or context.DeadlineExceeded from SQS --- v2/manager.go | 27 +++++++++++++++++++++++---- 1 file changed, 23 insertions(+), 4 deletions(-) diff --git a/v2/manager.go b/v2/manager.go index f46b6e0..cb166f6 100644 --- a/v2/manager.go +++ b/v2/manager.go @@ -100,12 +100,30 @@ func (m *Manager) Delete(ctx context.Context, msg Message, err error) { ) if err != nil { - m.Report(err) + switch { + case errors.Is(err, context.Canceled): + // in this specific case, we may have done the work but the context passed to us was cancelled. + // we may or may not *have* to delete here. + m.Log(ctx, slog.LevelDebug, "not deleting a context.Canceled message", + slog.String("messageID", msg.ID), + slog.String("error", err.Error()), + ) + return + case errors.Is(err, context.DeadlineExceeded): + // do not delete from the queue if the deadline is exceeded. + // this needs to be reported as well, or it'll cause a problem + // for end users not seeing visibility or results. + m.Log(ctx, slog.LevelDebug, "not deleting a context.DeadlineExceeded message", + slog.String("messageID", msg.ID), + slog.String("error", err.Error()), + ) + m.Report(err) + return + default: + m.Report(err) + } } - // if err is a failure, we can choose if we retry here. - // TODO: retry strategy - if err := m.client.Delete(ctx, msg); err != nil { m.Log(ctx, slog.LevelError, "failed to delete message", slog.String("messageID", msg.ID), @@ -124,6 +142,7 @@ func (m *Manager) Fetch(ctx context.Context) error { for { select { case <-ctx.Done(): + m.Log(ctx, slog.LevelDebug, "exiting fetch instance") return ctx.Err() default: m.Log(ctx, slog.LevelDebug, "fetching messages") From f608c6a0294ecfa2c840c0b8b99eebb771ee655e Mon Sep 17 00:00:00 2001 From: Amelia Wylie Date: Thu, 1 Oct 2026 12:11:55 +0100 Subject: [PATCH 11/11] feat(formigo/v2): only sleep if fetchDelay is valid, and default to 0 --- v2/manager.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/v2/manager.go b/v2/manager.go index cb166f6..e4a742d 100644 --- a/v2/manager.go +++ b/v2/manager.go @@ -164,7 +164,9 @@ func (m *Manager) Fetch(ctx context.Context) error { // allow configuring a delay after fetching messages // this may help prevent excessive pressure on AWS. - time.Sleep(m.fetchDelay) + if m.fetchDelay > 0 { + time.Sleep(m.fetchDelay) + } } } } @@ -242,7 +244,7 @@ func NewManager(client Client, opts ...Option) *Manager { func NewDefaultManager(client Client) *Manager { return &Manager{ client: client, - fetchDelay: time.Millisecond * 500, + fetchDelay: 0, fetchConcurrency: 2, workerConcurrency: 20, deadline: time.Second * 30,