Four threads inside a job: the GVL, the connection pool, and one hot row

The measurement post ended with a model: a deal with N messages costs 0.3 s to list, 0.2 s of engine overhead, and 1.15 s per message, all serial, on one worker thread. The database sat at 11%. The phase was latency-bound.

The obvious lever is concurrency. The obvious concurrency in a Rails job is Ruby threads. And the obvious mistake is letting those threads near the database. This post is the design that got a four-fold gain per rich deal, the constraints that shaped it, and the smaller write cuts that made a wider queue lane safe at the same time.

What the GVL allows, and what it does not

Ruby's global VM lock means one thread runs Ruby code at a time. It does not mean one thread waits on a socket at a time: blocking I/O in the standard library — Net::HTTP reading a response, a sleep, a Postgres round-trip through the pg gem — releases the lock while it waits. Four threads each waiting on an HTTP response overlap almost perfectly.

So the question for any "threads inside a job" design is: what fraction of the thread's time is waiting versus computing? For a body fetch it is 90% waiting. The Ruby that runs per response — parse a JSON document, pick out a string — is milliseconds. The HTML sanitiser that runs when the message is written is heavier, and it stays on the job thread, single-file, on purpose.

Four threads, then, because the phase's gain is bounded by the source API's burst window, not by the CPU: the lowest tier allows twenty requests per two seconds, and four in flight per lane thread stays under that even when the lane itself is widened to four.

The rule: fetch threads never touch the database

The worker's connection pool is sized to exactly what the process can hold: web threads plus job threads plus two. There is no slack for four extra connections per running task. And a job thread already holds one for the whole job — the queue library takes a session-level advisory lock on the job row when it starts performing, which pins that thread's connection until it finishes, whatever the code inside does.

That rule decides the whole shape: reserve, fetch, then write. The job thread reserves API slots (the one thing that needs a row lock), the fetch threads do HTTP against those slots, the job thread joins them and writes everything in one transaction per deal.

BODY_POOL = 4

# One deal: list its mail, fetch every body BEFORE any transaction opens,
# then write in WINDOW-sized transactions with a savepoint per message.
def import_deal_mail(deal, source_deal_id, stats)
  @client.each_page("deals/#{source_deal_id}/mailMessages") do |page|
    msgs   = page[:data].map { |w| w["data"] || w }.select { |m| m["id"] }
    bodies = fetch_bodies(msgs)                      # HTTP, parallel, no DB
    msgs.each_slice(WINDOW) do |window|
      ActiveRecord::Base.transaction do              # one commit per window
        window.each do |msg|
          outcome =
            begin
              write_email(deal, msg, bodies[msg["id"].to_s])   # savepoint inside write_once
            rescue ActiveRecord::Deadlocked, ActiveRecord::SerializationFailure
              raise                                  # window gone; the TASK retries
            rescue StandardError => e
              failed += 1
              note!(level: "error", subject: msg["id"], message: Migration.failure_reason(e))
              :failed
            end
          bump(stats, outcome)
        end
      end
    end
  end
end

def fetch_bodies(msgs)
  return {} if @migration.dry_run

  ids = msgs.map { |m| m["id"].to_s }
  ids.each_slice(BODY_POOL).each_with_object({}) do |slice, bodies|
    @client.reserve_slots!(slice.size)               # row lock + any waiting: job thread only
    threads = slice.map do |id|
      Thread.new { Rails.application.executor.wrap { [id, fetch_body(id)] } }
            .tap { |t| t.report_on_exception = false }
    end
    # Join EVERY thread even when one raised, then surface the first error
    # here — a transient error retries the task exactly as it did inline.
    errors = []
    ActiveSupport::Dependencies.interlock.permit_concurrent_loads do
      threads.each do |t|
        id, body = t.value
        bodies[id] = body
      rescue StandardError => e
        errors << e
      end
    end
    raise errors.first if errors.any?
  end
end

# Runs on a fetch thread: lease: false, because the slot was reserved by
# the job thread. A body the token cannot see is nil; the message still
# imports with its headers and snippet.
def fetch_body(id)
  data = @client.get("mailbox/mailMessages/#{id}", { include_body: 1 }, cost: :get, lease: false)["data"]
  data.is_a?(Hash) ? data["body"].to_s.presence : nil
rescue Client::PermanentError
  nil
end

Three details that are easy to get wrong:

• Every thread is joined, even after one raises. Thread#value re-raises the thread's exception in the caller; if you map(&:value) and the first raises, the other three are abandoned mid-request. Collect, then raise the first.
• executor.wrap and permit_concurrent_loads. In production, code is eager-loaded and the autoload interlock is a no-op. In development and test, a bare thread that touches an autoloadable constant can deadlock against the parent, which holds the load lock while it waits. The pair above is the documented way to run threads under Rails; use it everywhere, because the deadlock only appears in the environment you test in.
• The fetch happens before the transaction. The window transaction opens after fetch_bodies returns. Nothing is in flight to the source while a transaction is open, and no row lock is held across a network call. This is the oldest rule in the Rails book and the easiest to break by refactoring a loop.

The guard for the ordering records events: with a stub client that logs each fetch and a spy on transaction, three fetches must precede the single window, and the window must hold exactly three savepoints:

expect(order.reject { |e| e == :savepoint }).to eq(%i[fetch fetch fetch window])
expect(order.count(:savepoint)).to eq(3)

The hot row: budget accounting off its tuple

The source meters requests per account, and the engine keeps one budget row per account. Before this work, that row was written on every API response — update_columns with whatever the rate-limit headers said — plus a row-locked lease every 25 requests. In the twenty-hour phase that was about 154,000 autocommits on one tuple, each one racing the other lane thread for the same row, for a number the next response overwrote anyway.

Two things had to change for the fetch threads to exist at all: the per-response write had to go (a fetch thread must not need a connection), and the lease had to be taken on the job thread before the parallel round.

The new shape keeps the headers in memory, under a mutex because four threads absorb responses against one instance, and writes the row only when the reading matters:

class MigrationBudget < ApplicationRecord
  PERSIST_EVERY  = 30.seconds
  HEADER_COLUMNS = %w[burst_capacity burst_used burst_reset_at budget_remaining exhausted_at budget_reset_at].freeze

  after_initialize do
    @mutex = Mutex.new
    @spent_delta = 0
    @daily_moved = false
    @persisted_at = Time.current
  end

  # In memory only — the fetch-thread path. No row lock, no connection.
  def absorb(headers, spent_units: 0)
    names = self.class.header_map(provider)
    @mutex.synchronize do
      @spent_delta += spent_units.to_i
      self.burst_capacity = limit  if (limit = int_header(headers, names[:limit]))&.positive?
      self.burst_used     = [burst_capacity - remaining, 0].max if (remaining = int_header(headers, names[:remaining]))
      self.burst_reset_at = Time.current + reset.seconds if (reset = int_header(headers, names[:reset]))&.positive?
      if (daily = int_header(headers, names[:long_remaining]))
        @daily_moved ||= daily != budget_remaining
        self.budget_remaining = daily
        self.exhausted_at     = daily.positive? ? nil : Time.current
      end
    end
    self
  end

  # ONE write, when it matters: a lease is about to lock and reload the row,
  # the daily counter moved, or PERSIST_EVERY has passed. Only what THIS
  # instance learned is written — the header columns it changed, and spent
  # as an increment — so a stale instance cannot undo another worker's lease.
  def persist!(force: false)
    sets = @mutex.synchronize do
      due   = force || @daily_moved || @persisted_at < PERSIST_EVERY.ago
      dirty = changed & HEADER_COLUMNS
      next unless due && (dirty.any? || @spent_delta.positive?)

      @daily_moved, @persisted_at = false, Time.current
      delta, @spent_delta = @spent_delta, 0
      clear_attribute_changes(dirty)
      dirty.to_h { |c| [c.to_sym, self[c]] }.merge(delta: delta)
    end
    return unless sets

    delta = sets.delete(:delta)
    assignments = sets.keys.map { |c| "#{c} = :#{c}" } + ["spent = COALESCE(spent, 0) + :delta", "updated_at = :now"]
    self.class.where(id: id).update_all([assignments.join(", "), sets.merge(delta: delta, now: Time.current)])
  end

  def lease!(want: 25)
    raise Exhausted.new(retry_at: budget_reset_at, provider: provider) if long_budget_gone?
    persist!(force: true)          # what the headers said reaches the row BEFORE the lock reloads it
    with_lock do
      reset_window_if_expired!
      granted = [want, burst_capacity - burst_used].min.clamp(0, want)
      if granted.positive?
        self.burst_used += granted
        save!
      end
      granted
    end
  end
end

Two subtleties in persist! came from tests failing, not from design. Writing the whole row back from an instance would clobber another worker's lease with this instance's stale burst_used; hence only dirty header columns, and spent as an increment rather than a value. And lease! must persist before with_lock, because with_lock reloads the row and would discard a reading that says the window is fuller than our own count — the spec for that is a lease that must grant 3, not 5, after the headers said 17 of 20 were gone.

The client gets one new flag and one new method:

def get(path, params = {}, cost: :list, lease: true)
  await_slot! if lease                 # sleeps or parks the task — job thread only
  response = perform_get(build_uri(path, params))
  if @budget
    @budget.absorb(response, spent_units: Provider.cost(cost))
    @budget.persist! if lease          # throttled; never from a fetch thread
  end
  JSON.parse(response.body)
end

def reserve_slots!(count) = count.times { await_slot! }

The rare exception to "no database from a fetch thread" is a 429: the client writes the exhaustion to the row from wherever it happens, once, and the task is parked. That path runs a few times a year. Making it thread-pure was not worth the code.

Three smaller cuts that add up

With the fetch threads holding nothing, the lane could be widened. Before doing that, three more writes per task went, so that four threads would load Postgres less than two did before.

The run heartbeat, an UPDATE on the run row per task, is read by the reaper's zombie sweep at thirty minutes. It now writes at most every thirty seconds:

HEARTBEAT_EVERY = 30.seconds

def heartbeat!(migration)
  return if migration.heartbeat_at && migration.heartbeat_at > HEARTBEAT_EVERY.ago
  migration.update_columns(heartbeat_at: Time.current, updated_at: Time.current)
end

The run log, a capped JSON array rewritten per failed row, is buffered per task and appended in one statement, flushing early past the cap.

Ten deals per task. One task per deal seeded 30,788 queue rows, each maintaining nineteen indexes and kept for seven days, plus a claim, a checkpoint, a finish, a heartbeat and a top-up per deal. Discovery now seeds one task per ten deals; the checkpoint offset is the deal index, so retry and resume keep their meaning, and a task seeded before the change carries a single deal id and runs unchanged.

What one deal writes now

For a four-message deal, before and after:

The three inserts per message remain — a ledger claim, a timeline row and the message itself, on tables with 26 and 32 indexes — and they are the product's design, not the migration's. What went away was everything that was per-response or per-task bookkeeping.

Measured on the first write-heavy run after widening the lane to four threads: connections peaked at 26 of a 181 ceiling, CPU at 26%, write IOPS at 133, and all three returned to baseline within a minute of the run ending. The lane could have gone wider. It didn't need to.

What I deliberately left alone

• Replacing the pinned connection. Rails 7.2+ returns connections between queries unless code leases one, and two call sites in the task path did lease. I was going to fix them — until I checked that the queue library pins the job thread's connection for the whole perform anyway. The edits would have been correct and pointless. The fetch threads are what matter, and they hold nothing.
• synchronous_commit = off. Seventeen commits at about 1.5 ms each is 25 ms of a five-second deal, and the window transaction removed most of them. Not worth a durability caveat.
• A wider fetch pool. Four in flight per lane thread, four lane threads: sixteen requests in flight against an eighty-per-window burst limit. The lowest plan tier allows twenty. Going wider would trade a faster phase for a customer on a small plan getting throttled by their own migration.

Next: the search that took eleven seconds, which turned out to be the same lesson — a join the planner cannot index — in a different part of the app.