Shoryuken Documentation

If you need to control how Shoryuken selects which queue to poll next, you can implement a custom polling strategy by subclassing Shoryuken::Polling::BaseStrategy.

To create a functional strategy, you must override the following abstract methods:

• next_queue: Returns a QueueConfiguration object representing the next queue to poll, or nil if no queues are available.
• messages_found(queue, count): Invoked after polling a queue. Use this to adjust polling behavior (e.g., pausing empty queues or adjusting weights) based on the number of messages retrieved.
• active_queues: Returns an array of queues that are currently active and available for polling.

Additionally, you can optionally override message_processed(queue) to perform actions after a message has been successfully processed.

class CustomStrategy < Shoryuken::Polling::BaseStrategy
  def initialize(queues)
    @queues = queues
  end

  def next_queue
    # Return next queue to poll
    @queues.sample
  end

  def messages_found(queue, count)
    # Handle result of polling
    logger.info "Found #{count} messages in #{queue}"
  end

  def active_queues
    # Return list of active queues
    @queues
  end
end

Shoryuken provides the Shoryuken::Middleware::Server::AutoExtendVisibility middleware to prevent messages from becoming visible to other consumers while they are still being processed.

To use this feature, your worker class must implement and return true for auto_visibility_timeout?.

Important Constraints:

• Batch Workers: Auto-extension is not supported for batch workers. If a batch is detected, the middleware will log a warning and skip extension.
• Queue Timeout Requirements: The SQS queue's visibility timeout must be long enough to allow for a scheduled extension. If the queue's visibility timeout is too short (specifically, if it is $\le 0$ or results in a non-positive interval based on EXTEND_UPFRONT_SECONDS), the middleware will log a warning and will not extend the visibility.
• Mechanism: The middleware uses a TimerTask to call change_visibility on the SQS message at regular intervals before the current timeout expires.

Shoryuken provides the ExponentialBackoffRetry server middleware to automatically adjust the SQS visibility timeout when a job fails. This allows for increasing delays between retries.

To use this, your worker class must:

1. Implement a class method exponential_backoff? that returns true.
2. Provide Shoryuken options via get_shoryuken_options containing a retry_intervals key.

Retry Interval Configuration The retry_intervals option can be configured in two ways:

• An Array of Integers: The middleware uses the index corresponding to the current attempt (based on SQS ApproximateReceiveCount). If the number of attempts exceeds the array size, it uses the last element in the array.
• A Proc/Callable: A object that responds to .call(attempts), where attempts is the current attempt number.

Important Behaviors:

• Non-retryable Exceptions: If an exception is listed in the non_retryable_exceptions option, the middleware will re-raise it immediately without applying a backoff. This allows other middleware (like NonRetryableException) to handle the message (e.g., by deleting it).
• Batch Workers: Exponential backoff is not supported for batch workers; if a batch is processed, the middleware will simply yield and allow the error to propagate normally.
• Visibility Timeout Cap: The calculated visibility timeout is automatically capped to ensure it does not exceed the SQS maximum (approximately 43,200 seconds minus processing time).

To ensure request-scoped context (such as current_user, tenant, or locale) flows from the code that enqueues an ActiveJob to the job's execution, use Shoryuken::ActiveJob::CurrentAttributes.persist.

This integration automatically serializes the attributes of your ActiveSupport::CurrentAttributes classes into the SQS message body during enqueueing and restores them before the job executes.

Note on Cleanup: Unlike Sidekiq, which relies on the Rails executor to reset attributes, Shoryuken performs a blanket reset of all registered CurrentAttributes classes after every job execution to prevent thread-local data leakage in the worker thread pool.

require 'shoryuken/active_job/current_attributes'

# Register a single class
Shoryuken::ActiveJob::CurrentAttributes.persist('MyApp::Current')

# Or register multiple classes
Shoryuken::ActiveJob::CurrentAttributes.persist('MyApp::Current', 'MyApp::RequestContext')