A checkpointed migration engine on GoodJob

Last month I built the engine that moves a customer's whole CRM — contacts, companies, deals, leads, activities, notes and email history — out of Pipedrive and into the product I work on. The first real customer had about 400,000 records and, as it turned out, 90,000 emails attached to their deals. The import ran for most of a day, survived two deploys and one stop, and finished with a ledger that could say for every source row what happened to it.

This is the architecture post of a series. The next one is the machine that runs it — processes, queues, scheduling, the three kinds of concurrency, and the five levels of batching nested inside one another. The later ones are about what went wrong (a ledger that poisoned itself, a retry that stranded a quarter of a million records, an email phase that took twenty hours) and what I measured before changing anything. This one is the shape of the thing, and why each part exists.

Everything is generalised. The product and the customer are not named, and the numbers are real.

Why not one job

The first version of this importer was one background job per run: fetch a page, write the rows, fetch the next page, for hours. It had the three problems every long job has.

• A deploy killed it. The queue's worker is restarted on every deploy, and a job that has been running for six hours does not survive that. Whatever it was doing is lost, and whatever it wrote is half there.
• It could not tell you where it was. Progress lived in the process. The UI showed "running" and a count that reset on retry.
• It could not be retried. Retrying the job meant starting from the first page again, re-fetching everything against a metered third-party budget.

The fix for all three is the same idea: make the unit of work small, make it own its position, and make every write idempotent. Then the run is the sum of its tasks, a dead worker costs you one task's worth of time, and "retry" means "run the tasks that did not finish".

The plan: phases in a DAG

The source's entities depend on each other. A deal points at a contact and a company; a note points at any of them; an email belongs to a deal. So the run is a plan of phases, each of which may only start when the phase before it has finished:

The plan is data, not code paths. Each source provider declares its phases as an ordered list of kind groups, and the planner seeds tasks for a phase only when it reaches it. That is what lets a run for "notes and activities only" have two phases instead of six.

The task: a cursor, a checkpoint, a claim

A task is a row: which run, which kind, which phase, a cursor that says where it starts, a processed_offset that says how far it got, and a status in pending → running → succeeded | failed | cancelled.

For a page-walking kind the cursor is a path and a start offset, and the source's pages are addressed arithmetically, so a phase can be seeded as several stripes — tasks that walk every Nth page — and each finishing task creates the next task in its stripe. That is how a 155,000-row entity is walked by four workers without any of them needing to know about the others.

Claiming a task is one fenced update:

class MigrationTask < ApplicationRecord
  # Two workers racing for the same task both issue this UPDATE. The
  # `status = 'pending'` predicate is the fence: exactly one of them
  # changes a row, and only that one proceeds.
  def claim!(worker:)
    self.class.where(id: id, status: "pending").update_all(
      status:     "running",
      attempts:   Arel.sql("attempts + 1"),
      locked_at:  Time.current,
      locked_by:  worker.to_s.first(100),
      started_at: Arel.sql("COALESCE(started_at, NOW())"),
      updated_at: Time.current
    ) == 1
  end

  def checkpoint!(offset:, stats:)
    update_columns(processed_offset: offset, stats: stats, updated_at: Time.current)
  end
end

The job that runs a task is short on purpose. It loads the task and the run, refuses to claim if the run is terminal or being cancelled, defers itself if the database is under pressure, claims, hands off to the provider's handler, and records the outcome:

class Migrations::TaskJob < ApplicationJob
  queue_as :migrations
  retry_on Provider::Client::TransientError, wait: :polynomially_longer, attempts: 8

  def perform(task_id, migration_id)
    task      = MigrationTask.find_by(id: task_id) or return
    migration = Migration.find_by(id: migration_id) or return
    return if migration.terminal?

    if migration.cancellation_requested?
      task.finish!("cancelled")
      return Planner.advance_phase!(migration, task.phase)
    end
    return self.class.set(wait: 30.seconds).perform_later(task_id, migration_id) if SystemPressure.high?
    return unless task.claim!(worker: "#{Socket.gethostname}:#{job_id}")

    begin
      Provider.handler(task: task, migration: migration).call
      task.finish!("succeeded")
    rescue Provider::Client::TransientError, ActiveRecord::Deadlocked => e
      task.release!(error: e.message) # back to pending, cursor intact; retry_on re-enqueues
      raise
    rescue StandardError => e
      task.finish!("failed", error: e.message)
    end

    heartbeat!(migration)
    Planner.top_up!(migration)
    Planner.advance_phase!(migration, task.phase)
  end
end

Two things in there matter more than they look. A transient error releases the task rather than failing it: the cursor is kept, so the retry resumes mid-page instead of rewriting the page. And a deadlock is treated as transient for the same reason — Postgres aborts the whole transaction, so it cannot be contained by a per-row savepoint, and the honest remedy is "run it again".

Writing a page: windows and savepoints

A source page is 500 rows and cannot be one transaction: Postgres caches at most 64 subtransaction ids per top-level transaction, and past that every visibility check falls back to a slower path. So a page is written in windows of 25 rows, one transaction each, with one savepoint per row so that a malformed row loses itself and not its window:

WINDOW = 25

def run_page
  rows = @client.get(path, start: cursor_start, limit: 500)["data"]
  rows.each_slice(WINDOW).with_index do |window, i|
    offset = i * WINDOW
    next if offset + window.length <= @task.processed_offset # committed earlier

    ActiveRecord::Base.transaction do
      window.each_with_index do |row, j|
        next if offset + j < @task.processed_offset
        outcome =
          begin
            ActiveRecord::Base.transaction(requires_new: true) { writer.call(kind, row) }
          rescue ActiveRecord::Deadlocked, ActiveRecord::SerializationFailure
            raise # the whole window is gone; the TASK retries with its cursor
          rescue StandardError => e
            failed += 1
            note!(level: "error", subject: row["id"], message: Migration.failure_reason(e))
            :failed
          end
        bump(stats, outcome)
      end
      @task.checkpoint!(offset: offset + window.length, stats: stats)
    end
  end
  report_sweep_failures(failed, total: attempted)
end

The checkpoint is written inside the window's transaction, so a checkpoint can never claim more than was committed. And the sweep reports its failure count at the end: per-row isolation is right, but a page whose every row failed must still fail the task rather than finishing green with the count buried in a JSON column.

The ledger: every source row, once

The centre of the engine is one table, migration_records, with a unique index on (tenant, source, kind, source_id). Every write goes through a claim on that index, in the same transaction as the write:

class MigrationRecord < ApplicationRecord
  # Claim a source row and write the local record in ONE transaction.
  # The block does the write; the claim goes LAST so that a write that
  # raises leaves no claim behind for the retry to trip over.
  def self.write_once(migration:, kind:, source_id:, action: "created", reason: nil)
    record = nil
    won    = false
    transaction(requires_new: true) do
      record = block_given? ? yield : nil
      claimed = connection.select_all(sanitize_sql_array([<<~SQL, {
        INSERT INTO migration_records
          (migration_id, tenant_id, source, kind, source_id, record_type, record_id, action, reason, created_at)
        VALUES (:migration_id, :tenant_id, :source, :kind, :source_id, :record_type, :record_id, :action, :reason, :now)
        ON CONFLICT (tenant_id, source, kind, source_id) DO UPDATE SET
          migration_id = EXCLUDED.migration_id, record_type = EXCLUDED.record_type,
          record_id    = EXCLUDED.record_id,    action      = EXCLUDED.action,
          reason       = EXCLUDED.reason,       created_at  = EXCLUDED.created_at
        WHERE migration_records.action = 'skipped'
        RETURNING id
      SQL
        migration_id: migration.id, tenant_id: migration.tenant_id, source: migration.source,
        kind: kind, source_id: source_id.to_s, record_type: record&.class&.name, record_id: record&.id,
        action: action, reason: reason, now: Time.current
      }]))
      raise ActiveRecord::Rollback if claimed.rows.empty?
      won = true
    end
    won ? record : nil
  end
end

Read the ON CONFLICT clause slowly, because it is the whole contract. A row that was created or matched is exclusive: a second run, an overlapping worker, a retried page — all of them find the slot taken, roll back their own write, and report the row as already here. A row that was only ever skipped is a placeholder: the next attempt that can do better takes it over. That second half took a production incident to get right; it has its own post.

The ledger is also the join table for everything that comes later. A note arrives from the source with a person id and a deal id; the writer resolves both through the ledger into local ids, in one query per page, and links the note to what it finds. If it finds nothing, the note is skipped as "not linked to any record" — a placeholder, waiting for the run that imports its parents.

The barrier: phases advance by a fenced update

The phase barrier is a single integer on the run, current_phase, and it only ever moves forward. Every task calls this when it finishes:

def advance_phase!(migration, phase)
  advanced = Migration.where(id: migration.id, current_phase: phase).where(
    "NOT EXISTS (SELECT 1 FROM migration_tasks
                 WHERE migration_id = ? AND phase = ? AND status IN ('pending','running'))",
    migration.id, phase
  ).update_all(current_phase: phase + 1, updated_at: Time.current)
  return false if advanced.zero?

  next_phase = phase + 1
  next_phase < phases_for(migration).length ? seed_phase!(migration, next_phase) : finalize!(migration)
  true
end

Two fences in one statement: current_phase = ? means only the phase the run is actually on can be advanced (a late task from an earlier phase cannot), and the NOT EXISTS means it advances only when nothing in the phase is left. Fifty tasks can finish in the same second; exactly one of them seeds the next phase.

Keeping the lane fair and full

The queue has a dedicated lane for migrations with a small number of threads. Two rules keep it fair between tenants and full for any one of them.

The planner keeps at most MAX_INFLIGHT of a run's tasks enqueued at a time, twice the lane's width, and refills to the deficit when a task finishes:

MAX_INFLIGHT = 8 # 2× the lane width

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

def enqueue_pending!(migration, phase = nil, limit: MAX_INFLIGHT)
  target = phase || migration.reload.current_phase
  MigrationTask.where(migration_id: migration.id, status: "pending", phase: target)
               .order(:id).limit(limit).pluck(:id)
               .each { |id| Migrations::TaskJob.perform_later(id, migration.id) }
end

Refilling to the deficit is the second version of this code. The first enqueued exactly one successor per finished task — one in, one out — and the pool decayed to a single running task within minutes. Why is in the measurement post.

Enqueueing a task twice is harmless, because the job claims-or-exits. Under- enqueueing stalls the run, which is why a reaper on a five-minute cron also tops up runs whose pool is below width, releases tasks whose worker died mid-page (a running task with a stale lock), and finalises runs whose worker stopped without reporting.

Admission: one live run per workspace

Two runs writing one workspace at once is how three overlapping imports produced 11,000 duplicate company groups on an early tenant. So a run is admitted only if no other run for that workspace is live:

def conflicting_run(migration)
  scope = Migration.where(tenant_id: migration.tenant_id, source: migration.source)
                   .where.not(id: migration.id)
                   .where.not(status: Migration::TERMINAL)
  scope.where(status: LIVE_STATUSES).or(scope.where.not(started_at: nil)).first
end

The controller answers 409 with the other run's name. The same check guards retry and resume — a stopped run resumed under a live sibling would be exactly the case the rule exists for.

Pacing against the source

The source meters requests two ways: a burst window per token, and a daily token budget per account where a listing call costs 20 and a single fetch costs 2. The engine keeps one budget row per source account — not per tenant, because two tenants importing one account genuinely share its allowance — and workers lease slots from it in chunks under a row lock, spend them locally, and report what the response headers said. A 429 parks the task until the budget resets rather than burning retry attempts on a wait measured in hours.

That row turned out to be the hottest row in the database, which is the subject of the threading post.

What the UI gets

Because tasks are rows, the run page is a rollup query, not a process introspection: batches done and in flight, per-entity counts of imported, already-here, skipped and failed, the phase the barrier is on, and a capped log of grouped reasons. Stop is a flag the job checks before each claim, so it lands in seconds. Retry re-arms the failed tasks and rewinds the barrier to where they live. Resume does the same for a stopped run's cancelled tasks. None of those needed the engine to change shape; they are queries and updates against the same rows.

What I'd keep, and what I'd do earlier

I'd keep all of it: the ledger with the claim in the write's transaction, the fenced barrier, windows of 25 with a savepoint per row, tasks that own a cursor. They are boring, and they are why a deploy in the middle of a twenty-hour run cost thirty minutes instead of a day.

What I'd do earlier is the measuring. The engine was correct before it was fast, and every hour I spent guessing at why a phase was slow was an hour a single grouped query would have saved. That's the next post.