Glue code that makes hotwire-spark and rails reloader work under Falcon + async-cable

local_dev_reloads_under_falcon.rb

# frozen_string_literal: true

# Glue code that makes hotwire-spark work under Falcon + async-cable.
#
# hotwire-spark assumes Puma + `rails server`:
#   - Hotwire::Spark.enabled? gates install on `defined?(Rails::Server)`,
#     which is only set by `rails s`. Falcon boots through its own CLI.
#   - The cable server is mounted through the router and upgrades via rack
#     hijack + nio4r, which does not cooperate with Falcon's fiber scheduler.
#
# We install spark manually and bridge its cable server through
# Async::Cable::Middleware so the websocket goes through async-websocket.
#
# Pinning the initializer with `after: "hotwire_spark.config"` ensures spark's
# engine initializer (which copies config.hotwire.spark onto module accessors
# like css_paths) has run before us, so we don't have to populate the watch
# paths by hand. Rails initializers only accept a single `after:` value, so
# we still prime ActionCable's cable.yml/logger ourselves -- those get set in
# `action_cable.set_configs` which orders independently of our anchor.
# `before: :build_middleware_stack` keeps us early enough to register
# middleware.
module HotwireSparkFalcon
  # Hotwire::Spark::Middleware stores the current request on @request, which
  # races under concurrent request handling (Falcon runs requests on fibers
  # with allow_concurrency = true). Mirror upstream PR #109 by passing the
  # request/helpers through as locals instead. Remove when that PR merges.
  module ThreadSafeMiddleware
    def call(env)
      status, headers, response = @app.call(env)

      request = ActionDispatch::Request.new(env)
      controller = request.controller_instance

      if controller && headers["Content-Type"]&.include?("text/html")
        helpers = controller.helpers
        html = +""
        response.each { |part| html << part }
        html = inject_spark_options(html, helpers)
        html = inject_spark_javascript(html, helpers)
        headers["Content-Length"] = html.bytesize.to_s
        response = [html]
      end

      [status, headers, response]
    end

    private

    def inject_spark_javascript(html, helpers)
      script_path = helpers.path_to_asset("hotwire_spark.js")
      tag = helpers.javascript_include_tag(script_path, defer: "")
      html.sub("</head>", "#{tag}</head>")
    end

    def inject_spark_options(html, helpers)
      options = [
        (helpers.tag.meta(name: "hotwire-spark:logging", content: "true") if Hotwire::Spark.logging),
        helpers.tag.meta(name: "hotwire-spark:html-reload-method", content: Hotwire::Spark.html_reload_method),
        helpers.tag.meta(name: "hotwire-spark:cable-server-path", content: Hotwire::Spark.cable_server_path)
      ].compact.join("\n")
      html.sub("</head>", "#{options}</head>")
    end
  end

  # ActiveSupport::Concurrency::ShareLock keys ownership on Thread.current and
  # uses MonitorMixin, which is thread-reentrant. Falcon serves many request
  # fibers from a single thread, so under stock Rails the reloader interlock
  # treats every fiber as the same owner: a reload fiber can acquire the
  # exclusive lock while another fiber still holds a share, then clears
  # constants mid-render. Symptoms include `undefined method 'to_model' for an
  # instance of ActiveStorage::VariantWithRecord` from `image_tag` in a view.
  #
  # FiberShareLock mirrors active_support/concurrency/share_lock.rb but rekeys
  # ownership on Fiber.current. MonitorMixin is retained: the only yield point
  # inside its synchronize blocks is `cv.wait_while`, which releases the
  # monitor and re-acquires it on signal. The synchronize blocks themselves
  # perform only in-memory mutation, so under cooperative fiber scheduling no
  # other fiber can interleave between entry and exit.
  class FiberShareLock
    include MonitorMixin

    def raw_state # :nodoc:
      synchronize do
        fibers = @sleeping.keys | @sharing.keys | @waiting.keys
        fibers |= [@exclusive_fiber] if @exclusive_fiber

        data = {}
        fibers.each do |fiber|
          purpose, compatible = @waiting[fiber]
          data[fiber] = {
            fiber: fiber,
            sharing: @sharing[fiber],
            exclusive: @exclusive_fiber == fiber,
            purpose: purpose,
            compatible: compatible,
            waiting: !!@waiting[fiber],
            sleeper: @sleeping[fiber]
          }
        end
        yield data
      end
    end

    def initialize
      super
      @cv = new_cond
      @sharing = Hash.new(0)
      @waiting = {}
      @sleeping = {}
      @exclusive_fiber = nil
      @exclusive_depth = 0
    end

    def start_exclusive(purpose: nil, compatible: [], no_wait: false)
      synchronize do
        unless @exclusive_fiber == Fiber.current
          if busy_for_exclusive?(purpose)
            return false if no_wait

            yield_shares(purpose: purpose, compatible: compatible, block_share: true) do
              wait_for(:start_exclusive) { busy_for_exclusive?(purpose) }
            end
          end
          @exclusive_fiber = Fiber.current
        end
        @exclusive_depth += 1
        true
      end
    end

    def stop_exclusive(compatible: [])
      synchronize do
        raise "invalid unlock" if @exclusive_fiber != Fiber.current

        @exclusive_depth -= 1
        if @exclusive_depth == 0
          @exclusive_fiber = nil

          if eligible_waiters?(compatible)
            yield_shares(compatible: compatible, block_share: true) do
              wait_for(:stop_exclusive) { @exclusive_fiber || eligible_waiters?(compatible) }
            end
          end
          @cv.broadcast
        end
      end
    end

    def start_sharing
      synchronize do
        if @sharing[Fiber.current] > 0 || @exclusive_fiber == Fiber.current
          # already holding a lock
        elsif @waiting[Fiber.current]
          wait_for(:start_sharing) { @exclusive_fiber }
        else
          wait_for(:start_sharing) { busy_for_sharing?(false) }
        end
        @sharing[Fiber.current] += 1
      end
    end

    def stop_sharing
      synchronize do
        if @sharing[Fiber.current] > 1
          @sharing[Fiber.current] -= 1
        else
          @sharing.delete Fiber.current
          @cv.broadcast
        end
      end
    end

    def exclusive(purpose: nil, compatible: [], after_compatible: [], no_wait: false)
      if start_exclusive(purpose: purpose, compatible: compatible, no_wait: no_wait)
        begin
          yield
        ensure
          stop_exclusive(compatible: after_compatible)
        end
      end
    end

    def sharing
      start_sharing
      begin
        yield
      ensure
        stop_sharing
      end
    end

    def yield_shares(purpose: nil, compatible: [], block_share: false)
      loose_shares = previous_wait = nil
      synchronize do
        if (loose_shares = @sharing.delete(Fiber.current))
          if (previous_wait = @waiting[Fiber.current])
            purpose = nil unless purpose == previous_wait[0]
            compatible &= previous_wait[1]
          end
          compatible |= [false] unless block_share
          @waiting[Fiber.current] = [purpose, compatible]
        end

        @cv.broadcast
      end

      begin
        yield
      ensure
        synchronize do
          wait_for(:yield_shares) { @exclusive_fiber && @exclusive_fiber != Fiber.current }

          if previous_wait
            @waiting[Fiber.current] = previous_wait
          else
            @waiting.delete Fiber.current
          end
          @sharing[Fiber.current] = loose_shares if loose_shares
        end
      end
    end

    private

    def busy_for_exclusive?(purpose)
      busy_for_sharing?(purpose) ||
        @sharing.size > ((@sharing[Fiber.current] > 0) ? 1 : 0)
    end

    def busy_for_sharing?(purpose)
      (@exclusive_fiber && @exclusive_fiber != Fiber.current) ||
        @waiting.any? { |f, (_, c)| f != Fiber.current && !c.include?(purpose) }
    end

    def eligible_waiters?(compatible)
      @waiting.any? { |f, (p, _)| compatible.include?(p) && @waiting.all? { |f2, (_, c2)| f == f2 || c2.include?(p) } }
    end

    def wait_for(method, &block)
      @sleeping[Fiber.current] = method
      @cv.wait_while(&block)
    ensure
      @sleeping.delete Fiber.current
    end
  end

  # Trailing-edge debounce: each event extends the quiet-window deadline, and
  # we only flush once no event has arrived for QUIET_WINDOW seconds. Keyed by
  # (action, path) so multiple writes to the same file collapse to one frame.
  module Debouncer
    QUIET_WINDOW = 0.1
    @mutex = Mutex.new
    @pending = {}
    @flush_at = nil
    @worker = nil

    class << self
      attr_reader :mutex, :pending
      attr_accessor :flush_at, :worker
    end

    def broadcast
      key = [action, send(:canonical_changed_path)]
      Debouncer.mutex.synchronize do
        Debouncer.pending[key] = self
        Debouncer.flush_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + QUIET_WINDOW
        return if Debouncer.worker&.alive?
        Debouncer.worker = Thread.new { Debouncer.run }
      end
    end

    def self.run
      loop do
        remaining = mutex.synchronize { flush_at - Process.clock_gettime(Process::CLOCK_MONOTONIC) }
        if remaining > 0
          sleep(remaining)
        else
          to_send = mutex.synchronize { pending.values.tap { pending.clear } }
          to_send.each { |change| change.__send__(:broadcast_reload_action) if change.send(:should_broadcast?) }
          break
        end
      end
    end
  end

  # ActionDispatch::Executor takes a reloader share via `start_running` and
  # only releases it when the response body's `close` is called (Rails wraps
  # the response body in a BodyProxy whose finaliser is `state.complete!`).
  # For a WebSocket upgrade under Falcon, the response body is a Writable
  # queue that gets iterated for the entire lifetime of the connection -- so
  # `close` never fires and the share never drops. With FiberShareLock
  # correctly excluding fibers, this means an open WebSocket permanently
  # holds a share, and any subsequent constant reload (which needs an
  # exclusive `:unload`) deadlocks. Under Puma the same code path works
  # because ActionCable's worker pool detaches the response from the rack
  # body chain. Pre-empt the leak by dropping the share before entering the
  # cable path; cable channels are reloaded through their own mechanism.
  class CableShareReleaser
    def initialize(app)
      @app = app
    end

    def call(env)
      if Async::WebSocket::Adapters::Rack.websocket?(env)
        ActiveSupport::Dependencies.interlock.done_running
      end
      @app.call(env)
    end
  end

  class Railtie < ::Rails::Railtie
    initializer "hotwire_spark_falcon.install",
      after: "hotwire_spark.config",
      before: :build_middleware_stack do |app|
      next unless Rails.env.development?
      next unless defined?(Hotwire::Spark) && defined?(Async::Cable::Middleware)
      next unless Hotwire::Spark.enabled

      ActionCable::Server::Base.config.cable ||= Rails.application.config_for(:cable).deep_symbolize_keys
      ActionCable::Server::Base.config.logger ||= Rails.logger

      # Async::Cable's loop lets exceptions from handle_incoming kill the
      # socket, whereas the standard ActionCable mount swallows recoverable
      # protocol errors via its worker pool. Rescue the duplicate-subscribe
      # case spark's client occasionally produces so the connection survives.
      connection_class = Class.new(ActionCable::Connection::Base) do
        rescue_from ActionCable::Connection::Subscriptions::AlreadySubscribedError do |error|
          logger&.warn("Hotwire::Spark ignoring duplicate subscribe: #{error.message}")
        end
      end

      Hotwire::Spark.install_into(app)
      Hotwire::Spark.cable_server.config.connection_class = -> { connection_class }

      # Swap the reloader interlock to a fiber-keyed ShareLock. Safe to do at
      # boot: nothing has acquired the lock yet. See FiberShareLock above.
      ActiveSupport::Dependencies.interlock.instance_variable_set(:@lock, FiberShareLock.new)

      # Build watchers and macOS fsevents commonly fire several writes for one
      # logical change (esbuild rewrite + sourcemap, atomic rename, etc).
      # Spark broadcasts each Listen tick, so the browser gets a flurry of
      # reload frames. Coalesce identical (action, path) broadcasts within a
      # short window.
      Hotwire::Spark::Change.prepend(Debouncer)
      Hotwire::Spark::Middleware.prepend(ThreadSafeMiddleware)

      app.config.middleware.insert_before(Async::Cable::Middleware, CableShareReleaser)
      app.config.middleware.use(
        Async::Cable::Middleware,
        path: Hotwire::Spark.cable_server_path,
        server: Hotwire::Spark.cable_server
      )
    end
  end
end