Bulk Import

Bulk Import: the job every team writes badly once - load N thousand rows into the database without melting it, and survive the crash that WILL happen at row 2,743. Three disciplines, all boring, all load-bearing: BATCH (one insert of 500 beats 500 inserts of one - the database's time is billed per round-trip), IDEMPOTENT UPSERT (re-running a batch must be a no-op, because at-least-once is the only delivery guarantee reality offers), and a JOURNALED CURSOR (completed batches …

source on github

a real captured run

BULK IMPORT (batch, upsert, journal - and the crash is included)

  monday 17:04 - the import runs, and at batch 6 of 10: power cut.
    plan status: partial_failure; rows landed so far: 3000 (batches 0-5, durably journaled)

  monday 17:11 - same command, re-run. no flags, no surgery:
    journal says 6 batches already done -> skipped; resumed at batch 6
    plan status: completed; rows in db: 5000; total batch calls: 10

  and the at-least-once drill: batch 0 re-delivered on purpose ->
    row count 5000 -> 5000 (upsert made the duplicate delivery a non-event)

  the arithmetic that pays the rent: 5000 rows in 10 round-trips
  instead of 5000 (the database bills per round-trip, not per row);
  a resume that skipped exactly the 6 finished batches because
  the CURSOR lives in a durable journal, not in a variable that
  died with the process; and an upsert so boring that re-delivery
  is a shrug - which is the entire trick, because at-least-once
  is the only delivery guarantee production has ever offered
  anyone. batch for the database, journal for the crash, upsert
  for the truth.

source

# frozen_string_literal: true

# Bulk Import: the job every team writes badly once - load N
# thousand rows into the database without melting it, and survive
# the crash that WILL happen at row 2,743. Three disciplines, all
# boring, all load-bearing: BATCH (one insert of 500 beats 500
# inserts of one - the database's time is billed per round-trip),
# IDEMPOTENT UPSERT (re-running a batch must be a no-op, because
# at-least-once is the only delivery guarantee reality offers), and
# a JOURNALED CURSOR (completed batches are recorded durably, so the
# resume skips exactly what finished and repeats nothing). The crash
# is included. The crash is always included.
#
#   bundle exec ruby examples/bulk_import.rb
#
# Runs offline; exits 1 unless the import survives its own crash
# with zero duplicates and zero lost rows.

require class="s">"bundler/setup"
require class="s">"agentic"
require class="s">"tmpdir"

Agentic.logger.level = class="y">:fatal

ROWS = 5_000
BATCH = 500
CSV_PATH = File.join(Dir.tmpdir, class="s">"agentic_import.csv")
JOURNAL_PATH = File.join(Dir.tmpdir, class="s">"agentic_import_journal.jsonl")

# The database: a hash with a meter on it. Every call costs a round-trip.
DB = {}
DB_CALLS = Hash.new(0)
def db_upsert_batch(rows)
  DB_CALLS[class="y">:batch_upsert] += 1
  rows.each { |row| DB[row[class="s">"id"]] = row } # upsert: keyed write, naturally idempotent
end

# Generate the file (in real life: the CSV marketing sent you at 5pm).
# Parsed by hand: the csv stdlib leaves the default gems in 3.4, and
# this file has no quoting to worry about - the census taught us well.
File.delete(CSV_PATH) if File.exist?(CSV_PATH)
File.open(CSV_PATH, class="s">"w") do |f|
  f.puts class="s">"id,email,plan"
  ROWS.times { |i| f.puts class="s">"u#{i},user#{i}@example.com,#{["freeclass="s">", "proclass="s">"][i % 2]}" }
end

def read_rows(path)
  lines = File.foreach(path).map { |line| line.strip.split(class="s">",") }
  header = lines.shift
  lines.map { |values| header.zip(values).to_h }
end

def run_import(sabotage_batch: nil)
  journal = Agentic:class="y">:ExecutionJournal.new(path: JOURNAL_PATH)
  done = Agentic:class="y">:ExecutionJournal.replay(path: JOURNAL_PATH).events
    .select { |e| e[class="y">:event] == class="s">"batch_done" }.map { |e| e[class="y">:description].to_i }

  orchestrator = Agentic:class="y">:PlanOrchestrator.new(concurrency_limit: 1, retry_policy: {max_retries: 0, retryable_errors: []})
  batches = read_rows(CSV_PATH).each_slice(BATCH).to_a
  previous = nil
  skipped = 0
  batches.each_with_index do |rows, index|
    if done.include?(index)
      skipped += 1
      next
    end
    task = Agentic:class="y">:Task.new(description: class="s">"batch #{index}", agent_spec: {class="s">"name" => class="s">"importer", class="s">"instructions" => class="s">"w"})
    orchestrator.add_task(task, previous ? [previous] : [], agent: ->(_t) {
      raise Agentic:class="y">:Errors:class="y">:LlmAuthenticationError, class="s">"power cut at batch #{index}" if index == sabotage_batch
      db_upsert_batch(rows)
      journal.record(class="y">:batch_done, description: index.to_s, rows: rows.size)
      class="y">:imported
    })
    previous = task
  end
  status = orchestrator.execute_plan.status
  [status, skipped, batches.size]
end

puts class="s">"BULK IMPORT (batch, upsert, journal - and the crash is included)"
puts
File.delete(JOURNAL_PATH) if File.exist?(JOURNAL_PATH)

status, _, total = run_import(sabotage_batch: 6)
puts class="s">"  monday 17:04 - the import runs, and at batch 6 of #{total}: power cut."
puts class="s">"    plan status: #{status}; rows landed so far: #{DB.size} (batches 0-5, durably journaled)"
puts

status2, skipped, = run_import
puts class="s">"  monday 17:11 - same command, re-run. no flags, no surgery:"
puts class="s">"    journal says #{skipped} batches already done -> skipped; resumed at batch 6"
puts class="s">"    plan status: #{status2}; rows in db: #{DB.size}; total batch calls: #{DB_CALLS[class="y">:batch_upsert]}"
puts

# At-least-once drill: re-deliver an already-imported batch on purpose
before = DB.size
db_upsert_batch(read_rows(CSV_PATH).first(BATCH))
puts class="s">"  and the at-least-once drill: batch 0 re-delivered on purpose ->"
puts class="s">"    row count #{before} -> #{DB.size} (upsert made the duplicate delivery a non-event)"
puts

failures = []
failures << class="s">"crash run should have failed" unless status == class="y">:partial_failure
failures << class="s">"resume didn't skip completed work" unless skipped == 6
failures << class="s">"rows lost or duplicated (#{DB.size})" unless DB.size == ROWS
failures << class="s">"batches re-imported (#{DB_CALLS[class="y">:batch_upsert]} calls)" unless DB_CALLS[class="y">:batch_upsert] == total + 1 # 10 real + 1 drill
failures << class="s">"resume didn't complete" unless status2 == class="y">:completed

puts class="s">"  the arithmetic that pays the rent: #{ROWS} rows in #{total} round-trips"
puts class="s">"  instead of #{ROWS} (the database bills per round-trip, not per row);"
puts class="s">"  a resume that skipped exactly the #{skipped} finished batches because"
puts class="s">"  the CURSOR lives in a durable journal, not in a variable that"
puts class="s">"  died with the process; and an upsert so boring that re-delivery"
puts class="s">"  is a shrug - which is the entire trick, because at-least-once"
puts class="s">"  is the only delivery guarantee production has ever offered"
puts class="s">"  anyone. batch for the database, journal for the crash, upsert"
puts class="s">"  for the truth."
exit(failures.empty? ? 0 : 1)