Workers, queues, and batches inside batches

The architecture post explained the engine's data model: phases, tasks, the ledger, the barrier. This one is about the machine that runs it. Which processes execute jobs. How a job gets from a perform_later to a thread. Who decides what runs next, and how the lane is kept both full and fair. Where the three kinds of concurrency live. And the part I get asked about most: the batching, which is not one batch size but five, nested.

Everything is generalised; the numbers are the real ones.

One host, two containers, one database

Two containers run on one small host. A web container runs Puma with two workers of three threads each. A jobs container runs the queue worker: one process, one GVL, and inside it a scheduler per queue, each with its own thread pool. The migration lane is one of those schedulers, four threads wide.

Five things in that picture shape everything below.

A lane is its own executor, not a share of a pool. The queue library runs as a single capsule: one process, one GVL, and a scheduler per queue with its own thread pool. Nine schedulers, 24 threads. Widening the migration lane from two to four adds two threads to one executor; it cannot borrow from critical when it is busy, and it cannot starve critical when it is not. That isolation is the reason a 400,000-record import is allowed to run at all on the same process as password-reset mail.

Three things wake a thread. A notifier holding a LISTEN connection, which dispatches within milliseconds of an enqueue; a poller every fifteen seconds, which is the safety net for a notification nobody was listening for; and a cron manager for recurring entries, one of which is the reaper. A job that misses its notify is late, never lost.

A claim is a database lock, not a state field. A thread takes a session-level advisory lock on the job row. If the process dies, the connection dies, and the database drops the lock — the job is claimable again with no cleanup pass and no "stuck in running" sweep. That property is the foundation the engine's own claim! builds on: the queue guarantees one worker per job, and the fenced task update guarantees one task per worker.

The web tier executes nothing. Both containers carry the same job-thread setting, and a resolver switches the web one to external mode when Puma is clustered. If I had assumed otherwise, "a four-thread lane" would have meant twelve threads across three processes and the fetch-thread arithmetic would have been wrong threefold. It took an hour to confirm, and the capacity post is that hour.

The queue is a table in the application's own database. No broker. That is what lets a task claim work, write rows and record its checkpoint in one transaction — the single most useful property in the whole design. The cost is that queue rows are a storage line item, and a migration produces hundreds of thousands of them.

The lane's width and the connection pool are tied together by a spec: the queue string's numbers must sum to the declared thread count, and the deploy files must announce that same count, because the pool is sized from it. Widening the lane was a change to two files that a test refused to let disagree.

How a task reaches a thread

The engine never runs a migration "in a job". It enqueues one job per task, with two integer arguments — the task id and the run id — and nothing else. No token, no payload, no state. Everything the job needs is in the task row, and the row is the truth if the job dies.

Migrations::TaskJob.perform_later(task.id, migration.id)   # a good_jobs row

That enqueue is a row and a NOTIFY. The notifier in the jobs process is listening, hands the job to the scheduler for its queue, and a free thread in that scheduler's pool takes the advisory lock and runs it — milliseconds, end to end, with nothing in between. If all four lane threads are busy the row simply waits, and it is picked up on the next notify or, failing that, by the poller within fifteen seconds.

The job's perform is then a claim-or-exit. Two workers can be handed the same task id (the planner over-enqueues on a race, by design); the fenced UPDATE … WHERE status = 'pending' lets exactly one of them proceed. The other exits in a millisecond. So there are two locks on the path, and they answer different questions: the queue's advisory lock decides which worker runs a job, and the task's fenced update decides whether that worker has any work to do. That is what makes every scheduling path below safe to run concurrently: enqueueing twice is harmless, enqueueing zero times is the only failure, and the reaper exists to prevent that one.

Scheduling: who enqueues what, and when

Five things put work on the lane. They are independent, they overlap, and each is idempotent.

Start seeds the first phase and enqueues up to the in-flight limit. Seeding a page-walking kind creates several stripe tasks — one per starting page, striding by the stripe count — so four threads walk the collection from four places at once, and each finishing task creates the next task in its own stripe:

STRIPES = 4

def seed_rows_for(migration, phase, kind)
  Providers.for(migration.source)
           .seed_cursors(kind, stripes: STRIPES)      # ["organizations:0", "organizations:500", …]
           .map { |cursor_key, cursor| { migration_id: migration.id, phase:, kind:, cursor_key:, cursor: } }
end

def enqueue_next_stripe(path, start)
  next_start = start + (STRIPES * PAGE_SIZE)
  MigrationTask.create_with(phase: @task.phase, cursor: { "path" => path, "start" => next_start })
               .find_or_create_by!(migration_id: @migration.id, kind: @task.kind,
                                   cursor_key: "#{path}:#{next_start}")
end

find_or_create_by! against a unique index on (migration, kind, cursor_key) is what makes a retried page harmless: it cannot create a second successor.

Top-up runs when any task finishes, and refills to the deficit. The in-flight limit is per run — twice the lane's width — and it is the fairness control: two customers migrating at once each keep eight tasks queued and share the four threads roughly evenly, oldest first.

MAX_INFLIGHT = 8

def top_up!(migration)
  deficit = MAX_INFLIGHT - MigrationTask.where(migration_id: migration.id, status: "running").count
  enqueue_pending!(migration, limit: deficit) if deficit.positive?
end

Advance is the fenced barrier from the architecture post: a finishing task tries to close its phase, exactly one succeeds, and that one seeds the next phase. Everything else — the reaper's refill, enqueue_pending!, the retry path — is gated on the barrier's phase, so no task from a later phase can run before its dependencies exist.

The reaper is a five-minute cron with four sweeps, each a query: release running tasks whose lock is older than thirty minutes (a worker died mid-page; the cursor is intact, so the retry resumes), refill runs whose pool is below width, advance phases that closed without a winner, and finalise runs whose heartbeat stopped without a terminal status. It never decides anything the tasks do not already say; it only repeats what a finishing task would have done.

Pressure sits in front of the claim. If the database reports more active queries than a threshold, the job re-enqueues itself with a backoff without claiming, so the cursor is untouched and the task stays honestly pending.

Multithreading: three kinds, three places

There are three layers of concurrency, and each has a different job.

1. Lane threads — four, in the jobs process, each running one task at a time. This is throughput across tasks. They share the process's connection pool, each holding one connection for the duration of its job, and they share one GVL: four threads in one process interleave, they do not run on four cores. That is acceptable only because a migration task spends nearly all of its life waiting — on a socket, or on Postgres — and both waits release the GVL. A CPU-bound job in this lane would serialise the other three.
2. Fetch threads — four per running email task, created inside the job thread, doing HTTP only. This is latency hiding inside a task: Ruby's GVL is released during socket waits, so four body fetches overlap almost perfectly, and they hold no database connection because the pool has no room for them. The threads post is the whole design; the rule is reserve, fetch, then write.
3. Stripes — concurrency across a collection, not across threads: a collection walked from four starting pages at once is four tasks that any threads can pick up. It is why a two-thread lane still walked a 155,000-row collection at full width.

The three compose in the obvious way and the arithmetic is worth saying out loud: four lane threads, each able to run an email task, each of which runs four fetch threads, is sixteen concurrent HTTP requests against the source and four concurrent database transactions. The first number is why the rate limiter is leased centrally rather than per task; the second is why the connection pool never noticed.

What is deliberately not concurrent: writes. Every Postgres write for a task happens on the job thread, in windows, in order. The database's contention story is simple because only one thread per task ever holds a transaction.

Batches inside batches

"What's the batch size?" has five answers, because the batching nests. Each level owns a cursor or an offset, and each level is the unit of something:

Reading it from the outside in:

• Run — the unit of admission. One live run per workspace, enforced before any task exists.
• Phase — the unit of ordering. Closed by the fenced barrier; seeds the next.
• Task — the unit of scheduling and retry. It owns a cursor (which page, or which deals) and a processed_offset (how far), and it is what the queue hands to a thread. A retry re-runs a task from its offset, never a run from the start.
• Page — the unit of fetching: 500 rows, the source API's maximum, one listing call at 20 tokens regardless of size. There is nothing to gain from a smaller page and 20 tokens to lose.
• Window — the unit of atomicity: 25 rows, one transaction, the checkpoint written inside it. Twenty-five because each row opens two savepoints (the writer's and the ledger's) and Postgres caches 64 subtransaction ids per transaction; 50 stays under the line with the same margin the rest of the codebase uses. Bigger windows mean fewer commits and slower rollbacks; this size was chosen for the rollback.
• Row — the unit of isolation: one savepoint, so a malformed row is rolled back alone and counted, and the window commits without it.

The email task nests differently because its work is a tree, not a list:

• Task — ten deals, seeded by discovery. The offset is the deal index, so a stopped task resumes at the deal it was on. Ten because one task per deal had seeded 30,788 queue rows and paid the per-task bookkeeping 30,788 times.
• Deal — one listing call, then a checkpoint after the deal's messages are written. Nothing is fetched twice on resume, because the ledger recognises every message already imported.
• Page of messages — the listing's page.
• Fetch round — four bodies: reserve four rate-limit slots on the job thread, run four threads, join them all, repeat. No database.
• Window — 25 messages, one transaction, after every body on the page is in hand. Nothing is in flight to the source while a transaction is open.
• Message — one savepoint: the ledger claim and two inserts.

The two hierarchies share the bottom three levels on purpose. The window code is the same code; the email task just arrives at it with bodies already fetched.

How it was built, in order

The order mattered more than any single piece, so here it is.

1. One job per run, the version that existed. It proved the writers and the field mapping and nothing else.
2. Tasks with cursors, the ledger, the barrier. Correctness first: a deploy or a stop must cost one task, and every source row must be written once. Windows and savepoints came with this, sized from Postgres's subtransaction limit rather than tuned.
3. The two silent bugs, found by grouping the ledger and reading the run's stats rather than its log — a skip that claimed its slot, and a retry that stranded work behind the barrier.
4. Measurement. One query grouping task duration by the task's own payload said the email phase was latency-bound, and ranked every lever before any was pulled.
5. Write cuts, then concurrency. The budget row off the hot path, one transaction per deal, heartbeat and log throttled, ten deals per task — and only then fetch threads and a four-thread lane, because more threads against the old write pattern would have doubled the contention rather than the throughput.
6. Capacity, then the lane. Connection ceiling, pool arithmetic, storage growth per run, the host's credit mode, the web tier's execution mode — read from the environment, not assumed from the code — before widening anything.

Each step left a guard: a spec that fails when the step is reverted, and that I reverted to watch fail. That is the discipline the whole series is really about; the engine is what it happened to produce.