Processing Events

[SteffanPerry]

Describe the feature

This gem works great for processing messages enqueued by the application. However one of the great features of SQS is the ability to listen to events from SNS (or other sources).

We currently are handling this by moving events into lambda functions, but we would like to stop using lambda for this.

Before i start, looking for some feedback or suggestions.

Use Case

This gem doesn't provide a clear way on how to process these items. Specifically we have an application comprised of 7 rails services each emitting events and processing them in other services. For example the account service may emit an event when an a new account has been created, another service may listen to this event and do something.

Proposed Solution

Im working on a fork for us to use which will allow something like:

Config:

queues:
  production_default_queue: 
    url: 'https://sqs.us-east-1.amazonaws.com/000000000/production-default-queue'
  production_external_event_queue:
    url: 'https://sqs.us-east-1.amazonaws.com/000000000/production-external-events-queue' # Subscribed to some SNS topic
    handler: MyEventJob

Job:

class MyEventJob
  def perform
      do_stuff(@event)
  end
end

Other Information

No response

Acknowledgements

• I may be able to implement this feature request
• This feature might incur a breaking change

SDK version used

latest

Environment details (OS name and version, etc.)

all

[alextwoods]

I assume by "Events" you mean some arbitrary message that follows some well defined schema per queue? IE: the queue itself here is what determines what Job class should be called.

What parameters would you want your job class to be called with?

I like the general idea of this, but am worried about conflating the ActiveJob logic with arbitrary/un-modeled event processing. In the above configuration for example, what should the behavior be if a job is sent to the production_default_queue?

[alextwoods]

One thought might be to have configuration for an "unknown job handler" (name tbd) - similar to what you have above, but the idea would be that if the poller receives a message that it can't map to a known job that it will call the configured handler. This could be both a mechanism for handling non-active job events as well as a potential error handling mechanism.

[SteffanPerry]

As for the parameters of what the class would be called with, i think it should be the raw SQS message. Im using "event" a synonym for SQS message here.

These queues i think should be separate from a a standard job queue. regarding your question about the production_default_queue, "events" should not be sent to this queue, rather a 1:1 queue to job mapping. ie all messages in this queue will always and only be processed by the defined job.

Although this is not exactly how ActiveJob currently works (by this I mean the producer of the event knowing the consumer and consumers class) this is how many event based applications work built with SQS/SNS, and is the only real missing piece to use it for non monolithic applications.

The only issue i see with the "unknown job handler" is that is limits you to a single job. If i have more than one job processing different events, now i need to build some sort of routing within a single job.

[mullermp]

Perhaps not immediately useful, but have you looked at Kafka for what you are trying to accomplish?

[SteffanPerry]

I have used kafka in the past, but to be honest, my opinion is that SQS/SNS might be the single greatest product AWS has ever released. additionally it's much more affordable (generally) to use and requires no maintenance / management. Adding an additional item to our stack is not something I'm too interested in doing at the moment unless it's absolutely necessary.

[SteffanPerry]

@alextwoods i actually just saw that i used the term "production_default_queue" as my example, that was copy pasta mistake. i have updated it to be more clear.

[alextwoods]

Although this is not exactly how ActiveJob currently works (by this I mean the producer of the event knowing the consumer and consumers class)

Yeah - I think this is my major concern here. Certainly I think we could add this in fairly simply, but I worry about conflating the two models (activejob vs message processing).

Does the Aws::SQS::QueuePoller in the SDK meet the needs? (the polling in this gem is built on top of it).

[alextwoods]

The only issue i see with the "unknown job handler" is that is limits you to a single job.

If I was understanding your suggested config above, the handler limits you to a single job class per queue. I think the unknown job handler config I was suggesting would also be configured per queue.

[SteffanPerry]

If I was understanding your suggested config above, the handler limits you to a single job class per queue. I think the unknown job handler config I was suggesting would also be configured per queue.

This makes sense.

But I also can see your point about separating true active job "jobs" from message events. The poller will work fine with some additional work. I was trying to find a "all in one" solutions but maybe thats not realistic currently.

[alextwoods]

I'm not saying we definitely shouldn't support message events - I do think theres utility in the config/poller from this gem that would be useful for this case and I also really love sqs/sns as a mechanism in complex apps where you don't control the many sources of events/messages.

I just want to make sure that if we do support it, we do so in a way that doesn't make usage confusing.

[SteffanPerry]

I have been doing some testing in production with one of my less critical services. I was able to modify just the job runner class to handle the events and convert them into active job events. thoughts on this possible option?

Config:

queues:
  development_event_test_queue:
    url: 'https://sqs.us-east-1.amazonaws.com/000000000000/development-service-event-test'
    job_class: MyEventJob

Example Job:

class MyEventJob < ApplicationJob
  def perform(sqs_message)
    puts "performing my event job: #{sqs_message}"
  end
end

JobRunner.rb

module Aws
  module ActiveJob
    module SQS
      # @api private
      class JobRunner
        attr_reader :id, :class_name

        def initialize(message)
          @job_data   = prepare_job_data(message)
          @class_name = @job_data['job_class'].constantize
          @id         = @job_data['job_id']
        end

        def self.queue_event_handlers
          @@queue_handlers ||= {}.tap do |handlers|
            Aws::ActiveJob::SQS.config.queues.values.each do |queue_config|
              next unless queue_config[:job_class].present?

              handlers[queue_config[:url]] = queue_config[:job_class]
            end
          end
        end

       .....existing code....

        private

        def prepare_job_data(message)
          data = ActiveSupport::JSON.load(message.data.body)
          return data if data['job_class'].present?

          prepare_event_data(message)
        rescue JSON::ParserError
          prepare_event_data(message)
        end

        def prepare_event_data(message)
          {
            'job_class' => job_class_from_config(message.queue_url),
            'job_id' => message.data[:message_id],
            'arguments' => [message.data.to_json]
          }
        end

        def job_class_from_config(queue_url)
          return queue_event_handlers[queue_url] if queue_event_handlers[queue_url]

          raise ArgumentError, "No handler configured for queue #{queue_url}"
        end

        def queue_event_handlers
          self.class.queue_event_handlers
        end
      end
    end
  end
end

[alextwoods]

I think the general idea of this makes sense - and I like the idea of falling back to some configured class/unknown_job_handler/event_handler (not sure on naming).

I guess one question - for handling these events, is using an ActiveJob class the correct way to handle these events? I think its worth thinking through how this interacts with other ActiveJob features like exceptions/retry or callbacks.

What happens with your code above when an exception is raised during processing of the event?

[SteffanPerry]

@alextwoods From my point of view, and i am biased so take that into consideration, I like the idea of a single job processor. Most rails devs are use to active job, and expect it to work a certain way. so IMO, this solution is the closest to how traditional rails applications work. That is, you have an API or Front end layer (web request), and a worker layer (Active jobs). Adding a third background worker to process events, while surely doable, takes away from the developer happiness of rails.

Theoretically, all we have done here is changed the enqueueing mechanism, even though its invoked on the processing side. Since we are converting the message to the default ActiveJob format. everything going forward should work exactly as if it was enqueued via active job directly.

For example:
an event comes in, its a sqs message that does not meet the expected AJ format:

{
    'job_class' => ...,
    'job_id' => ...,
    'arguments' => [...]
}

We fallback and convert it IF we have a configured queue. then process it, if you notice i have not altered the "run" method because this message is now in the expected format.

If the job fails, there are 3 main scenarios (taken from the readme):

1 Default retry_on
In general - you should configure retries for your jobs using [retry_on](https://api.rubyonrails.org/classes/ActiveJob/Exceptions/ClassMethods.html#method-i-retry_on). When configured, ActiveJob will catch the exception and reschedule the job for re-execution after the configured delay. This will delete the original message from the SQS queue and requeue a new message.

Theoretically, the message will be re-enqueued in the activejob format. In this scenario when that new message is reprocessed by rails, it will be in the correct activejob format and return and process correctly from this guard:

def prepare_job_data(message)
          data = ActiveSupport::JSON.load(message.data.body)
          return data if data['job_class'].present?

2 Job left on queue:
By default, any StandardError raised during job execution will leave the message on the queue (retrying it later) and initiate shutdown for the poller and it will attempt to finish executing any in progress jobs.

In this scenario, the job should just reprocess in the exact same way (converting to AJ format before calling run).

3 Custom error handling
You can configure an error handler block for the ActiveJob SQS poller with poller_error_handler. Jobs that raise a StandardError and that do not handle that error via retry_on or discard_on will call the configured poller_error_handler with |exception, sqs_message|. You may re-raise the exception to terminate the poller. You may also choose whether to delete the sqs_message or not. If the message is not explicitly deleted then the message will be left on the queue and will be retried following the SQS Queue's configured [retry and DLQ settings](https://docs.aws.amazon.com/lambda/latest/operatorguide/sqs-retries.html). Retries provided by this mechanism are after any retries configured on the job with retry_on.

In this scenario, we are asking the dev to handle the raw SQS message manually. This is possibly the only corner case. You could argue that the developer expects it to be in the active job format here, so if they try to do something such as: job_class = JSON.parse(sqs_message.data.body)['job_class'], this could cause issues.

The counter point to this is that the devs owns this application and should know these messages could be different and handle it accordingly

That leads me to one other possible solution, though I don't know if i'm a huge fan of it. That would be to alter the run method here, so instead of converting to the correct format and invoking the job, we fetch the correct job class as we are doing above, and enqueue a new message, then delete the original event. The benefit here would be handling the errors in scenario 3 above.