The Attachment Pipeline

The Attachment Pipeline: Shrine's central lesson is that file uploads are a TWO-PHASE commit wearing a file input - phase one (cache) must be instant and disposable, phase two (promote + derivatives) is slow, background, and idempotent, because users double-submit, workers die mid-thumbnail, and retries must never double-bill. A plan with a journal is exactly the right machine for phase two.

source on github

a real captured run

THE ATTACHMENT PIPELINE (cache instantly, promote carefully)

  phase 1 (request): cached team-photo.jpg as upload-7f3a - 0ms of processing

  phase 2, attempt 1 (background): 3 derivatives scheduled...
    worker crashed at derive:web:1200 - status: partial_failure
    journal holds 2 paid derivative(s): derive:thumb:200, derive:ocr_text
    record NOT promoted; cache still serves the original. users see a photo, not an error.

  phase 2, attempt 2 (retry): only 1 derivative(s) scheduled - the paid ones skipped
    promoted: true; cache cleared: true

  phase 2, attempt 3 (the double-submit): 0 derivatives scheduled, nothing re-derived,
    promotion already recorded - idempotent all the way down.

  the shape to steal: CACHE is cheap and lies to nobody (the user's
  file is safe the instant the request returns); PROMOTION is a
  journaled plan whose derivative names are idempotency keys, so a
  crash resumes at the exact thumbnail it died on and a retry
  re-derives NOTHING. promotion commits the record only after every
  derivative exists - the record is the two-phase commit's second
  phase. uploads look like a file input; they're a distributed
  transaction, and pretending otherwise is where the corrupted
  avatars come from.

source

# frozen_string_literal: true

# The Attachment Pipeline: Shrine's central lesson is that file
# uploads are a TWO-PHASE commit wearing a file input - phase one
# (cache) must be instant and disposable, phase two (promote +
# derivatives) is slow, background, and idempotent, because users
# double-submit, workers die mid-thumbnail, and retries must never
# double-bill. A plan with a journal is exactly the right machine
# for phase two.
#
#   bundle exec ruby examples/attachment_pipeline.rb
#
# Runs offline; the "upload" is a hash, the crash is real.

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

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

STORES = {cache: {}, store: {}}
UPLOAD = {id: class="s">"upload-7f3a", filename: class="s">"team-photo.jpg", bytes: 48_213}.freeze

# Phase 1 - cache: instant, no processing, happens in the request
STORES[class="y">:cache][UPLOAD[class="y">:id]] = UPLOAD
puts class="s">"THE ATTACHMENT PIPELINE (cache instantly, promote carefully)"
puts
puts class="s">"  phase 1 (request): cached #{UPLOAD[class="y">:filename]} as #{UPLOAD[class="y">:id]} - 0ms of processing"
puts

# Phase 2 - promotion: a journaled plan. Derivative names are the
# idempotency keys, so a crashed promotion resumes instead of re-paying.
JOURNAL_PATH = File.join(Dir.tmpdir, class="s">"agentic_promote_#{UPLOAD[class="y">:id]}.jsonl")
File.delete(JOURNAL_PATH) if File.exist?(JOURNAL_PATH)

DERIVATIVES = {
  class="s">"derive:thumb:200" => 0.02,
  class="s">"derive:web:1200" => 0.03,
  class="s">"derive:ocr_text" => 0.04
}.freeze

def promotion_plan(upload, crash_at: nil)
  journal = Agentic:class="y">:ExecutionJournal.new(path: JOURNAL_PATH)
  done = Agentic:class="y">:ExecutionJournal.replay(path: JOURNAL_PATH)
  orchestrator = Agentic:class="y">:PlanOrchestrator.new(
    concurrency_limit: 2, lifecycle_hooks: journal.lifecycle_hooks,
    retry_policy: {max_retries: 0, retryable_errors: []}
  )

  derivative_tasks = DERIVATIVES.filter_map do |name, cost|
    next if done.completed?(name) # already paid for - skip, don't re-derive

    task = Agentic:class="y">:Task.new(description: name, agent_spec: {class="s">"name" => name, class="s">"instructions" => class="s">"derive"})
    orchestrator.add_task(task, agent: ->(_t) {
      raise class="s">"worker OOM-killed" if crash_at == name

      sleep(cost)
      class="s">"#{name.split(":class="s">")[1]} of #{upload[class="y">:filename]}"
    })
    task
  end

  unless done.completed?(class="s">"promote:record")
    promote = Agentic:class="y">:Task.new(description: class="s">"promote:record", agent_spec: {class="s">"name" => class="s">"promote", class="s">"instructions" => class="s">"p"})
    orchestrator.add_task(promote, derivative_tasks, agent: ->(_t) {
      STORES[class="y">:store][upload[class="y">:id]] = upload.merge(promoted: true)
      STORES[class="y">:cache].delete(upload[class="y">:id])
      class="s">"promoted"
    })
  end
  [orchestrator, derivative_tasks.size]
end

# First attempt: the worker dies mid-derivatives
orchestrator, scheduled = promotion_plan(UPLOAD, crash_at: class="s">"derive:web:1200")
result = orchestrator.execute_plan
state = Agentic:class="y">:ExecutionJournal.replay(path: JOURNAL_PATH)
puts class="s">"  phase 2, attempt 1 (background): #{scheduled} derivatives scheduled..."
puts class="s">"    worker crashed at derive:web:1200 - status: #{result.status}"
puts class="s">"    journal holds #{state.completed_descriptions.size} paid derivative(s): #{state.completed_descriptions.join(", class="s">")}"
puts class="s">"    record NOT promoted; cache still serves the original. users see a photo, not an error."
puts

# The retry (double-submitted by an anxious user AND the job system)
orchestrator, scheduled = promotion_plan(UPLOAD)
orchestrator.execute_plan
puts class="s">"  phase 2, attempt 2 (retry): only #{scheduled} derivative(s) scheduled - the paid ones skipped"
puts class="s">"    promoted: #{STORES[class="y">:store].key?(UPLOAD[class="y">:id])}; cache cleared: #{!STORES[class="y">:cache].key?(UPLOAD[class="y">:id])}"
puts

third, scheduled = promotion_plan(UPLOAD)
third.execute_plan
puts class="s">"  phase 2, attempt 3 (the double-submit): #{scheduled} derivatives scheduled, nothing re-derived,"
puts class="s">"    promotion already recorded - idempotent all the way down."
puts
puts class="s">"  the shape to steal: CACHE is cheap and lies to nobody (the user's"
puts class="s">"  file is safe the instant the request returns); PROMOTION is a"
puts class="s">"  journaled plan whose derivative names are idempotency keys, so a"
puts class="s">"  crash resumes at the exact thumbnail it died on and a retry"
puts class="s">"  re-derives NOTHING. promotion commits the record only after every"
puts class="s">"  derivative exists - the record is the two-phase commit's second"
puts class="s">"  phase. uploads look like a file input; they're a distributed"
puts class="s">"  transaction, and pretending otherwise is where the corrupted"
puts class="s">"  avatars come from."