Tenant Shards

Tenant Shards: at scale, "the plan" becomes "the plan, per shard" - same pipeline, isolated blast radius. Each shard gets its own journal (its own recovery story) and its own rate limit (its own noisy neighbor containment), under one control plane. Shard 2 crashes mid-run; the rerun resumes ONLY what shard 2 didn't finish, because recovery, like everything at scale, must be per-shard.

source on github

a real captured run

TENANT SHARDS (3 shards, 6 tenants, pipeline: extract -> transform -> load)

  run 1 - shard_2 loses power mid-tenant:
    shard_1   completed         6 steps ran
    shard_2   partial_failure   7 steps ran  <- crashed at umbrella:transform
    shard_3   completed         3 steps ran

  run 2 - control plane reruns everything; journals decide what that means:
    shard_1   completed         0 steps ran,  6 skipped (already journaled)
    shard_2   completed         2 steps ran,  7 skipped (already journaled)
    shard_3   completed         0 steps ran,  3 skipped (already journaled)

  the rerun was issued fleet-wide - the control plane doesn't
  track which shard crashed, and shouldn't have to. shard 1 and 3
  skipped everything (their journals proved completion); shard 2
  re-ran only from the crash point. that's the sharding contract:
  one plan definition, N isolated executions, N recovery stories,
  N rate limits - and a blast radius that ends at the shard
  boundary. scale isn't a bigger machine; it's smaller failures.

source

# frozen_string_literal: true

# Tenant Shards: at scale, "the plan" becomes "the plan, per shard" -
# same pipeline, isolated blast radius. Each shard gets its own
# journal (its own recovery story) and its own rate limit (its own
# noisy neighbor containment), under one control plane. Shard 2
# crashes mid-run; the rerun resumes ONLY what shard 2 didn't finish,
# because recovery, like everything at scale, must be per-shard.
#
#   bundle exec ruby examples/tenant_shards.rb
#
# Runs offline; the crash is scripted, the resume is real.

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

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

SHARDS = {
  class="s">"shard_1" => %w[acme globex],
  class="s">"shard_2" => %w[initech umbrella hooli],
  class="s">"shard_3" => %w[wonka]
}.freeze
PIPELINE = %w[extract transform load].freeze

def journal_path(shard) = File.join(Dir.tmpdir, class="s">"agentic_#{shard}.journal.jsonl")

# One shard's run: its own journal, its own limiter, resume-aware
def run_shard(shard, tenants, crash_at: nil)
  journal = Agentic:class="y">:ExecutionJournal.new(path: journal_path(shard))
  done = Agentic:class="y">:ExecutionJournal.replay(path: journal_path(shard))
  limiter = Agentic:class="y">:RateLimit.new(2) # per-shard: a hot shard can't starve the others

  orchestrator = Agentic:class="y">:PlanOrchestrator.new(
    concurrency_limit: 2, lifecycle_hooks: journal.lifecycle_hooks,
    retry_policy: {max_retries: 0, retryable_errors: []}
  )
  ran = 0
  skipped = 0
  tenants.each do |tenant|
    previous = nil
    PIPELINE.each do |step|
      name = class="s">"#{tenant}:#{step}"
      if done.completed?(name)
        skipped += 1
        next
      end

      task = Agentic:class="y">:Task.new(description: name, agent_spec: {class="s">"name" => name, class="s">"instructions" => class="s">"run"})
      orchestrator.add_task(task, previous ? [previous] : [], agent: ->(t) {
        raise class="s">"power cut" if crash_at == t.description

        limiter.acquire { sleep(0.002) }
        ran += 1
        class="y">:ok
      })
      previous = task
    end
  end
  status = orchestrator.execute_plan.status
  [status, ran, skipped]
end

puts class="s">"TENANT SHARDS (#{SHARDS.size} shards, #{SHARDS.values.sum(&class="y">:size)} tenants, pipeline: #{PIPELINE.join(" -> class="s">")})"
puts
SHARDS.each_key { |shard| File.delete(journal_path(shard)) if File.exist?(journal_path(shard)) }

puts class="s">"  run 1 - shard_2 loses power mid-tenant:"
SHARDS.each do |shard, tenants|
  crash = (shard == class="s">"shard_2") ? class="s">"umbrella:transform" : nil
  status, ran, = run_shard(shard, tenants, crash_at: crash)
  puts format(class="s">"    %-9s %-16s %2d steps ran%s", shard, status, ran,
    crash ? class="s">"  <- crashed at #{crash}" : class="s">"")
end
puts

puts class="s">"  run 2 - control plane reruns everything; journals decide what that means:"
SHARDS.each do |shard, tenants|
  status, ran, skipped = run_shard(shard, tenants)
  puts format(class="s">"    %-9s %-16s %2d steps ran, %2d skipped (already journaled)", shard, status, ran, skipped)
end
puts
puts class="s">"  the rerun was issued fleet-wide - the control plane doesn't"
puts class="s">"  track which shard crashed, and shouldn't have to. shard 1 and 3"
puts class="s">"  skipped everything (their journals proved completion); shard 2"
puts class="s">"  re-ran only from the crash point. that's the sharding contract:"
puts class="s">"  one plan definition, N isolated executions, N recovery stories,"
puts class="s">"  N rate limits - and a blast radius that ends at the shard"
puts class="s">"  boundary. scale isn't a bigger machine; it's smaller failures."