Async Redis key mutation notifications in Rails

Update 10/7/2022: I prepared a simple Rails 7 app that demonstrates the functionality described in this article.

---

I am a huge fan of Kredis. It allows Rails developers to see Redis as far more than just a fragment cache and "where jobs are".

Working with Kredis made me want to be able to run arbitrary operations in my Rails app when specific keys are modified via specific Redis commands. Redis has an excellent pub/sub infrastructure, and all Redis commands publish messages.

Why would someone want this?

While it's true that changes to data that occur within a typical Rails app are already well covered by model callbacks, state machines and other standard tooling, an entire world of real-time stream processing, ETL and multi-application use cases open up when you can run redis-cli set leastbad rules on your terminal and pick it up in your app.

Problem #1: Listening for messages blocks execution.
Solution #1: Spin up a thread!

Problem #2: Every dyno/server is going to receive the same messages, causing mayhem as developers respond to those messages with database updates. Side-effect chaos!
Solution #2: A standalone process that can be registered as a worker in Procfile... sort of like Sidekiq.

At first, I was just planning on borrowing 95% of Mike Perham's battle-hardened code. Then I realized that the Venn diagram of "people who want a Redis changeset firehose" and "Sidekiq users" is close to 100%... so I just bolted what I needed onto Sidekiq.

Try it out!

What follows is the MVP of my new gem. In fact, it's not a gem, yet: it's an initializer! It has no tests and is hours old. My janky code would make poor Mike bleed out. The goal is to see if folks actually need/want this to exist. I'm looking for feedback on what the ideal Rails-side API would actually look like.

Your Rails app needs to be up and running with Sidekiq. Just stick this in config/initializers/sidekiq.rb:

module Sidekiq
  class Subscriber
    include ::Sidekiq::Util

    def initialize
      @done = false
      @thread = nil
    end

    def start
      @thread ||= safe_thread("subscriber") {
        until @done
          Sidekiq.redis do |conn|
            # https://redis.io/topics/notifications#configuration
            conn.config(:set, "notify-keyspace-events", "E$lshz")
            # https://redis.io/topics/notifications#events-generated-by-different-commands
            conn.psubscribe("__key*__:*") do |on|
              on.psubscribe do
                @firehose = Firehose.new
              end
              on.pmessage do |pattern, command, key|
                @firehose.process(command.split(":").last.to_sym, key)
              end
              on.punsubscribe do
                @firehose = nil
              end
            end
          end
        end
        Sidekiq.logger.info("Subscriber exiting...")
      }
    end

    def terminate
      @done = true
      if @thread
        t = @thread
        Thread.kill(@thread)
        @thread = nil
        t.value
      end
    end
  end
end

module CoreExtensions
  module Sidekiq
    module Launcher
      attr_accessor :subscriber

      def initialize(options)
        @subscriber = ::Sidekiq::Subscriber.new
        super(options)
      end

      def run
        super
        subscriber.start
      end

      def quiet
        subscriber.terminate
        super
      end

      def stop
        subscriber.terminate
        super
      end
    end
  end
end

Sidekiq.configure_server do
  require "sidekiq/launcher"
  ::Sidekiq::Launcher.prepend(CoreExtensions::Sidekiq::Launcher)
end

I'm using CableReady to send console log notifications to the Console Inspector whenever a key is updated with the Redis SET command. I have a simple AllUsers ActionCable Channel in play for testing. This lives in app/lib/firehose.rb:

class Firehose
  include CableReady::Broadcaster

  attr_reader :redis

  def initialize
    @redis = ::ActionCable.server.pubsub.redis_connection_for_subscriptions
  end

  def process(command, key)
    case command # https://github.com/rails/kredis#examples
    when :set    # string, integer, json
      cable_ready["all_users"].console_log(message: "#{key} was just updated to #{redis.get(key)}").broadcast
    when :rpush  # list
    when :lrem   # unique_list
    when :sadd   # set
    when :incr
    when :decr
    when :incrby
    when :decrby
    when :exists
    when :del
      cable_ready["all_users"].console_log(message: "#{key} was deleted").broadcast
    else
    end
  end
end

As Seinfeld would say, is this anything?

Comments

[Vladimir Dementyev]

TIL Redis Keyspace Notifications.

That looks interesting.

I can't say I'm convinced by the redis-cli set whatever argument, but multi-application or cross-service scenarios sound reasonable.

Do you have a particular use-case in mind?

[leastbad]

You named the use cases!

You could have two monoliths sharing one Redis database. You could be dealing with a microservices setup. It could even just be multiple instances of the same app or dyno talking to a shared Redis instance.

Big picture, I think it's super weird and sad how our community only sees Redis as a glorified memcached upgrade. It has an embedded Lua engine! It can do realtime ML tricks on a video feed. It's outrageously flexible and powerful. It's like buying a PC and then only running WinAmp on it. Sure, music is cool, but...

For example, you could set up a Redis instance to process cryptocurrency data in realtime, or web server logs, or IoT sensor data. I'm pretty sure you could bring in OpenStreetMap data and run a route planner.