Class: Shoryuken::Middleware::Server::NonRetryableException

Inherits:
Object
  • Object
show all
Includes:
Util
Defined in:
lib/shoryuken/middleware/server/non_retryable_exception.rb

Overview

Middleware that handles non-retryable exceptions by deleting messages immediately. When a configured exception occurs, the message is deleted instead of being retried.

Configure non-retryable exceptions per worker:

class MyWorker
include Shoryuken::Worker

# Using exception classes
shoryuken_options queue: 'my_queue',
                  non_retryable_exceptions: [InvalidInputError, RecordNotFoundError]

# Or using a lambda for dynamic classification
shoryuken_options queue: 'my_queue',
                  non_retryable_exceptions: ->(error) {
                    error.is_a?(StandardError) && error.message.include?('permanent')
                  }

def perform(sqs_msg, body)
  # ...
end
end

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Util

#elapsed, #fire_event, #logger, #unparse_queues, #worker_name

Class Method Details

.non_retryable?(exception, non_retryable_exceptions) ⇒ Boolean

Checks whether an exception is classified as non-retryable for a worker

Parameters:

  • exception (Exception) —

    the exception to classify

  • non_retryable_exceptions (Array<Class>, #call, nil) —

    the worker's non_retryable_exceptions option: exception classes or a callable

Returns:

  • (Boolean) —

    true if the exception must not be retried



37
38
39
40
41
42
43
44
45
 
# File 'lib/shoryuken/middleware/server/non_retryable_exception.rb', line 37

def self.non_retryable?(exception, non_retryable_exceptions)
  return false unless non_retryable_exceptions

  if non_retryable_exceptions.respond_to?(:call)
    !!non_retryable_exceptions.call(exception)
  else
    Array(non_retryable_exceptions).any? { |klass| exception.is_a?(klass) }
  end
end

Instance Method Details

#call(worker, queue, sqs_msg, _body) { ... } ⇒ void

This method returns an undefined value.

Processes a message and handles non-retryable exceptions

Parameters:

  • worker (Object) —

    the worker instance

  • queue (String) —

    the queue name

  • sqs_msg (Shoryuken::Message, Array<Shoryuken::Message>) —

    the message or batch

  • _body (Object) —

    the parsed message body (unused)

Yields:

  • continues to the next middleware in the chain



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/shoryuken/middleware/server/non_retryable_exception.rb', line 55

def call(worker, queue, sqs_msg, _body)
  yield
rescue => e
  non_retryable_exceptions = worker.class.get_shoryuken_options['non_retryable_exceptions']

  return raise unless self.class.non_retryable?(e, non_retryable_exceptions)

  # Handle batch messages
  messages = sqs_msg.is_a?(Array) ? sqs_msg : [sqs_msg]

  logger.warn do
    "Non-retryable exception #{e.class} occurred for message(s) #{messages.map(&:message_id).join(', ')}. " \
    "Deleting message(s) immediately. Error: #{e.message}"
  end

  logger.debug { e.backtrace.join("\n") } if e.backtrace

  # Delete the message(s) immediately
  entries = messages.map.with_index { |message, i| { id: i.to_s, receipt_handle: message.receipt_handle } }

  begin
    queue_client = Shoryuken::Client.queues(queue)
    delete_failed = queue_client.delete_messages(entries: entries)

    # Check if deletion reported failures (returns true if any failed)
    if delete_failed
      logger.warn do
        'Failed to delete some messages for non-retryable exception on queue ' \
          "'#{queue}'. " \
          "Entries: #{entries.map { |e| { id: e[:id] } }.inspect}. " \
          'Some messages may remain in the queue and could be reprocessed.'
      end
    end
  rescue => delete_error
    logger.error do
      'Error deleting messages for non-retryable exception on queue ' \
        "'#{queue}': #{delete_error.class} - #{delete_error.message}. " \
        "Entries: #{entries.map { |e| { id: e[:id] } }.inspect}. " \
        'Messages may remain in the queue and could be reprocessed.'
    end
    logger.debug { delete_error.backtrace.join("\n") } if delete_error.backtrace
  end

  # Don't re-raise - the exception has been handled by deleting the message
end