4.1.0 writes broadcasts late when the broadcasting thread stays busy between them #92

# A thread that stays busy between broadcasts holds the GVL, so solid_cable
# 4.1.0's BatchedBroadcaster writer thread can't write the queued broadcasts
# until that thread finishes. A job streaming an LLM reply behaves like this:
# it renders a partial for every chunk. This script prints when each row was
# written, taken from its created_at, compared with when it was broadcast.
#
#   createdb solid_cable_repro
#   DATABASE_URL=postgres:///solid_cable_repro SOLID_CABLE_VERSION=4.1.0 ruby solid_cable_busy_broadcaster.rb
#   DATABASE_URL=postgres:///solid_cable_repro SOLID_CABLE_VERSION=4.0.2 ruby solid_cable_busy_broadcaster.rb

require "bundler/inline"

gemfile(true) do
  source "https://rubygems.org"
  gem "rails", "~> 8.1.0"
  gem "pg"
  gem "solid_cable", ENV.fetch("SOLID_CABLE_VERSION", "4.1.0"), require: false
end

require "rails"
require "active_record/railtie"
require "active_job/railtie"
require "action_cable/engine"
require "solid_cable"
require "tmpdir"
require "fileutils"

root = Dir.mktmpdir
FileUtils.mkdir_p(File.join(root, "config"))
File.write(File.join(root, "config/cable.yml"), <<~YAML)
  development:
    adapter: solid_cable
    polling_interval: 0.1.seconds
    message_retention: 1.day
YAML

class ReproApp < Rails::Application
  config.load_defaults 8.1
  config.eager_load = true
  config.logger = Logger.new(nil)
  config.secret_key_base = "repro"
end
ReproApp.config.root = root
Rails.application.initialize!

ActiveRecord::Base.establish_connection(ENV.fetch("DATABASE_URL"))
ActiveRecord::Schema.verbose = false
ActiveRecord::Schema.define do
  create_table "solid_cable_messages", force: :cascade do |t|
    t.binary "channel", limit: 1024, null: false
    t.binary "payload", limit: 536870912, null: false
    t.datetime "created_at", null: false
    t.integer "channel_hash", limit: 8, null: false
    t.index ["channel"]
    t.index ["channel_hash"]
    t.index ["created_at"]
  end
end

cable_config = ActionCable::Server::Configuration.new
cable_config.cable = { "adapter" => "solid_cable" }
cable_config.logger = Logger.new(nil)
pubsub = ActionCable::Server::Base.new(config: cable_config).pubsub

sent = []
Thread.new do
  Rails.application.executor.wrap do
    20.times do |i|
      pubsub.broadcast("stream", "chunk #{i}")
      sent << Time.current
      # Stands in for rendering the next chunk: 80 ms of Ruby work, GVL held.
      finish = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 0.08
      nil while Process.clock_gettime(Process::CLOCK_MONOTONIC) < finish
    end
  end
end.join
finished = Time.current
sleep 1 # let the writer catch up

written = SolidCable::Message.where(channel: "stream").order(:id).pluck(:created_at)
lags = written.each_with_index.map { |at, i| ((at - sent[i]) * 1000).round }
puts "ruby #{RUBY_VERSION}, rails #{Rails.version}, solid_cable #{SolidCable::VERSION}"
puts "written before the broadcasting thread finished: #{written.count { |at| at < finished }}/#{sent.size}"
puts "write delay per broadcast (ms): #{lags.inspect}"
pubsub.shutdown