Sending event id to Sidekiq instead of serialized event

[matiasgarcia]

In the docs, there is some suggested code to manage async handlers on Sidekiq.

The Scheduler is as follows:

module RailsEventStore
  class SidekiqScheduler
    def call(klass, serialized_event)
      klass.perform_async(serialized_event.to_h)
    end

    def verify(subscriber)
      subscriber.is_a?(Class) && subscriber < Sidekiq::Worker
    end
  end
end

And the AsyncHandler from the library is:

module RailsEventStore
  module AsyncHandler
    def perform(payload)
      super(Rails.configuration.event_store.deserialize(payload.symbolize_keys))
    end
  end
end

This works perfectly fine but we found ourselves hitting max memory error on Redis under heavy load. This is when I realized that the Scheduler serializes the whole event.

I was wondering if it's possible to provide just the event_id and write a custom AsyncHandler to retrieve it.

Like this:

module RailsEventStore
  class SidekiqScheduler
    def call(klass, serialized_event)
      klass.perform_async(serialized_event.event_id)
    end

    def verify(subscriber)
      subscriber.is_a?(Class) && subscriber < Sidekiq::Worker
    end
  end
end

module RailsEventStore
  module AsyncHandler
    def perform(payload)
      event = RailsEventStoreActiveRecord::Event.find(payload)
      super(event)
    end
  end
end

Or is there any problem on doing so?

[mostlyobvious]

I was wondering if it's possible to provide just the event_id and write a custom AsyncHandler to retrieve it.

That sounds reasonable and totally valid approach as long as it is scheduled "after commit" (so the event is already committed and visible outside any possible transaction wrapping publish process).

Nitpicks:

• RailsEventStoreActiveRecord::Event is a private detail of an adapter, you shouldn't use it directly — there's an API on Client for retrieving event by its id https://railseventstore.org/docs/read/#reading-specific-events
• I'd prefer having your app specific namespace to redefining modules and classes from RailsEventStore namespace for scheduler and handler behaving differently than an upstream one, you can swap these components in configuration:

event_store = RailsEventStore::Client.new(
  dispatcher: RubyEventStore::ComposedDispatcher.new(
                     RubyEventStore::ImmediateAsyncDispatcher.new(scheduler: MyScheduler.new),
                     RubyEventStore::Dispatcher.new)
)

[matiasgarcia]

Thanks for the guidance @pawelpacana!

Our rails event store is configured as follows:

  event_store = RailsEventStore::Client.new(
    dispatcher: RubyEventStore::ComposedDispatcher.new(
      RailsEventStore::AfterCommitAsyncDispatcher.new(scheduler: RailsEventStore::SidekiqScheduler.new),
      RubyEventStore::Dispatcher.new
    )
  )

Could it be that with RailsEventStore::AfterCommitAsyncDispatcher I don't really have to publish events in an after_commit hook but instead after_save/update/destroy should be enough?

[mpraglowski]

There are two additional callbacks that are triggered by the completion of a database transaction: after_commit and after_rollback. These callbacks are very similar to the after_save callback except that they don't execute until after database changes have either been committed or rolled back. They are most useful when your active record models need to interact with external systems which are not part of the database transaction.

from: https://guides.rubyonrails.org/active_record_callbacks.html#transaction-callbacks

The "external systems which are not part of the database transaction" the documentation mentions here is the Redis used by Sidekiq to store scheduled jobs.

Could it be that with RailsEventStore::AfterCommitAsyncDispatcher I don't really have to publish events in an after_commit hook but instead after_save/update/destroy should be enough?

Then you are at risk of running handlers (Sidekiq jobs) for events that have not been stored because transaction has been rolled back.

[mostlyobvious]

@matiasgarcia with RailsEventStore::AfterCommitAsyncDispatcher you're fine — the dispatcher would not schedule anything before commit or after rollback

ApplicationRecord.transaction do
   # do something
   event_store.publish(...) # not scheduling yet, as others might not see uncommitted changes from this transaction or it might be rolledback
end
# scheduling now

[matiasgarcia]

Thanks @pawelpacana.

I was wondering if maybe a maintained version of this AsyncHandler should be in the gem? I am saving 92% of my Redis storage with this change.

[mostlyobvious]

It's an interesting tradeoff to save on redis memory in exchange for additional query on a database. I'm not sure yet what is better as a default choice in AsyncHandler.

I'd accept PR with and alternative or configurable strategy for AsyncHandler for now 👍

[paneq]

@pawelpacana I like the idea!

I think we could change the implementation but keep the dispatcher/scheduler contract of (event, serialized_event) the same so that if someone wants to use the serialized version they can still do it. It's beneficial in case the consumer does not have access to RES storage.

[matiasgarcia]

It's an interesting tradeoff to save on Redis memory in exchange for an additional query on a database. I'm not sure yet what is better as a default choice in AsyncHandler.

I'd accept PR with and alternative or configurable strategy for AsyncHandler for now

In our case, we were firing around 200k events in a short time span (10 minutes) with a Redis DB of 100MB storage. This made it quickly reach the maximum threshold, because in our events we were tracking changes to AR models, making the payload quite big. So in this specific use case, it paid off to just send the event id. The additional query to fetch the event should be fairly quick since they are ids, although I don't have a benchmark to guarantee this :)

[mostlyobvious]

For the record and anyone looking for clear transition path:

• following handler mixin works well both with full payload and just an event_id

module EventFromEventId
  def perform(payload)
    event_id = payload.symbolize_keys.fetch(:event_id)
    super(event_store.read.event(event_id))
  end
end

• passing slice of the event payload containing event_id, for one-way compatibility

class Scheduler
  def call(klass, serialized_event)
    klass.perform_later({ event_id: serialized_event.event_id })
  end
  def verify(subscriber)
    Class === subscriber && !!(subscriber < ActiveJob::Base)
  end
end