[Rails World 2026] Durable orchestration on Rails: from continuation to workflow

Rails 8.1 shipped Active Job Continuation so long-running jobs survive restarts and resume from a checkpoint. This was a real step forward, but can only be the beginning. The workflows Rails apps orchestrate today need more than interruptibility: durable state across steps, pause/resume capabilities, and human-in-the-loop-like interactions. Building complex workflows and agentic pipelines in Rails has been simplified but still requires manual plumbing or reaching out to third-party gems or services.

This talk explores the preexisting patterns and solutions to the durable workflow orchestration problem and introduces a small, principled extension to Active Job that connects the dot: same step DSL, same job runner, but now with persistent context, scheduled steps, and first-class resumeability.

Transcript

1. FROM CONTINUATION TO WORKFLOW Durable orchestration on Rails Vladimir Dementyev Evil Martians
2. Vladimir on Rails Building: since 2014
3. Vladimir on Rails Building: since 2014 Contributing: since 2015
4. Vladimir on Rails Building: since 2014 Contributing: since 2015 ↳ 30+ apps, 30+ PRs, 30 gems, 1 book
5. Vladimir on Rails 8.2 Action Cable: - custom servers (Async) - fixes - features - performance
6. Vladimir on Rails 8.2 Action Cable: - custom servers (Async) - fixes - features - performance That proposal hasn't been selected :(
7. evilmartians.com
8. evilmartians.com/clients
9. rubyonrails.org/ai
10. rubyonrails.org/ai I see load-bearing slop
11. MY JOB
12. YOUR JOBS
13. YOUR BACKGROUND JOBS
14. ACTIVE JOB: FROM 4.2 TO 8.2
15. ACTIVE JOB
16. ACTIVE JOB 4.2: A common interface on top of existing gems 1.6k 2014
17. ACTIVE JOB 2.3k 4.2: A common interface on top of existing gems 1.6k 5.1: retry_on / discard_on 2014 2017
18. ACTIVE JOB 3.6k 2.3k 4.2: A common interface on top of existing gems 1.6k 7.1: perform_all_later 5.1: retry_on / discard_on 2014 2017 2023
19. ACTIVE JOB 7.2: enqueue_after_transaction_commit 3.8k 3.6k 2.3k 4.2: A common interface on top of existing gems 1.6k 7.1: perform_all_later 8.0: solid_queue by default 5.1: retry_on / discard_on 2014 2017 2023 2024
20. ACTIVE JOB 8.1: Continuation 7.2: enqueue_after_transaction_commit 4.7k 3.8k 3.6k 2.3k 4.2: A common interface on top of existing gems 1.6k 7.1: perform_all_later 8.0: solid_queue by default 5.1: retry_on / discard_on 2014 2017 2023 2024 2025
21. CONTINUATION 101 Interruptable and resumable jobs # app/jobs/event/webhook_dispatch_job.rb @fizzy class Event::WebhookDispatchJob < ApplicationJob include ActiveJob::Continuable def perform(event) step :dispatch do |step| Webhook.active.triggered_by(event) .find_each(start: step.cursor) do |webhook| webhook.trigger(event) step.advance! from: webhook.id end end end end
22. CONTINUATION 101 Interruptable and resumable jobs Steps and cursors as checkpoints # app/jobs/event/webhook_dispatch_job.rb @fizzy class Event::WebhookDispatchJob < ApplicationJob include ActiveJob::Continuable def perform(event) step :dispatch do |step| Webhook.active.triggered_by(event) .find_each(start: step.cursor) do |webhook| webhook.trigger(event) step.advance! from: webhook.id end end end end
23. CONTINUATION 101 Interruptable and resumable jobs Steps and cursors as checkpoints # app/jobs/event/webhook_dispatch_job.rb @fizzy class Event::WebhookDispatchJob < ApplicationJob include ActiveJob::Continuable def perform(event) step :dispatch do |step| Webhook.active.triggered_by(event) .find_each(start: step.cursor) do |webhook| Especiallywebhook.trigger(event) helpful when doingstep.advance! deploys from: webhook.id with Kamal end end end end
24. WITHOUT CONTINUATION class WebhookDispatchJob < ApplicationJob def perform(event) Webhook.active.triggered_by(event) .find_each do |webhook| webhook.trigger(event) end end end end
25. WITHOUT CONTINUATION class WebhookDispatchJob < ApplicationJob def perform(event) Webhook.active.triggered_by(event) .find_each do |webhook| webhook.trigger(event) end end end end Deploy
26. WITHOUT CONTINUATION
27. WITHOUT CONTINUATION class WebhookDispatchJob < ApplicationJob def perform(event) Webhook.active.triggered_by(event) .in_batches do |webhooks| webhooks.map do WebhookTriggerJob.new(event, it) end.then { ActiveJob.perform_all_later(it) } end end end class WebhookTriggerJob < ApplicationJob def perform(ev, webhook) = webhook.trigger(ev) end
28. WITHOUT CONTINUATION class WebhookDispatchJob < ApplicationJob class Event::WebhookDispatchJob < ApplicationJob def perform(event) include ActiveJob::Continuable Webhook.active.triggered_by(event) .in_batches do |webhooks| def perform(event) webhooks.map do step :dispatch do |step| WebhookTriggerJob.new(event, it) Webhook.active.triggered_by(event) end.then { ActiveJob.perform_all_later(it) } .find_each(start: step.cursor) do |webhook| end webhook.trigger(event) end step.advance! from: webhook.id end end end class WebhookTriggerJob < ApplicationJob end def perform(ev, webhook) = webhook.trigger(ev) end end
29. ACTIVE JOB 8.1: Continuation 7.2: enqueue_after_transaction_commit 4.7k 3.8k 3.6k 2.3k 4.2: A common interface on top of existing gems 1.6k 7.1: perform_all_later 8.0: solid_queue is default 5.1: retry_on / discard_on 2014 2017 2023 2024 2025
30. ACTIVE JOB FOUNDATION 4.7k 3.8k 3.6k 2.3k FACADE 1.6k 2014 2017 2023 2024 2025
31. ACTIVE JOB 4.9k 4.7k FOUNDATION 3.8k 3.6k 2.3k FACADE 1.6k 2014 2017 2023 2024 2025 2026 8.2: ...
32. ACTIVE JOB ON RAILS 8.2 Transactional integrity by default Better Continuations: attributes, typed cursors Debounce/throttling?
33. CONTINUATION IN THE WILD class SubmitEnrollmentJob < ApplicationJob include ActiveJob::Continuable attribute :token, :string attribute :profile_id, :integer Multi-step jobs w/ their own state def perform(enrollment) step :tokenize_payment do self.token = Payment.tokenize(enrollment.payment_instrument) end step :create_billing_profile do customer_id = enrollment.user_id self.profile_id = BillingProfile.create(customer_id:) end step :submit_enrollment do submission_id = EnrollmentAPI.submit( enrollment, token, profile_id) enrollment.update!(status: 'processing', submission_id:) end end end
34. CONTINUATION IN THE WILD Multi-step jobs w/ their own state Isolated steps (for fairness' sake) class CardGenerationJob < ApplicationJob include ActiveJob::Continuable def perform(card) @card = card step :moderate, isolated: true unless card.failed? step :generate, isolated: true end end end
35. CONTINUATION IN THE WILD Multi-step jobs w/ their own state Isolated steps (for fairness' sake) ↳ From infra to orchestra
36. ONE #step IS NOT ENOUGH #step
37. PATTERNS Pipelines
38. PATTERNS Pipelines: - imports, sync, backfills - media / docs processing - business flows (billing, enrollment) - provisioning
39. PIPELINES: CHAIN OF JOBS # some_controller.rb @sfruby-clouds PrepareImageJob.perfom_later(card) class PrepareImageJob < ApplicationJob def perform(card) card.preparing! card.image.variant(:web).processed card.prepared! AnalyzeCardJob.perform_later(card) end end
40. PIPELINES: CHAIN OF JOBS # some_controller.rb @sfruby-clouds PrepareImageJob.perfom_later(card) class PrepareImageJob < ApplicationJob class AnalyzeCardJob < ApplicationJob def perform(card) def perform(card) card.preparing! card.analyzing! card.image.variant(:web).processed if NSFWDetector.check(card.image.variant(:web)) card.prepared! card.analyzed! AnalyzeCardJob.perform_later(card) GenerateCardJob.perform_later(card) end else end card.fail!("NSFW check failed") end end end
41. PIPELINES: CHAIN OF JOBS class GenerateCardJob < ApplicationJob def perform(card) # some_controller.rb @sfruby-clouds card.generating! PrepareImageJob.perfom_later(card) card_generator = CardGenerator.new(cloud) io = card_generator.generate class PrepareImageJob < ApplicationJob class AnalyzeCardJob < ApplicationJob cloud.generated_image.attach(io:) def perform(card) def perform(card) card.generated! card.preparing! card.analyzing! end card.image.variant(:web).processed if NSFWDetector.check(card.image.variant(:web)) end card.prepared! card.analyzed! AnalyzeCardJob.perform_later(card) GenerateCardJob.perform_later(card) end else end card.fail!("NSFW check failed") end end end
42. PIPELINES: CHAIN OF STEPS class CardGenerationJob < ApplicationJob include ActiveJob::Continuable def perform(card) @card = card step :prepare do card.preparing! card.image.variant(:web).processed card.prepared! end step :moderate, isolated: true unless card.failed? step :generate, isolated: true unless card.failed? end end
43. PIPELINES: STATE MACHINE # app/models/cable/diagnostics.rb @anycable-plus class Cable::Diagnostic < ApplicationCachedRecord workflow do state :provider_status state :websocket_status state :admin_api_status state :completed after_transition :broadcast end performs def perform_next_step result = perform_check(workflow.current_state.name) workflow.next!(result.level) save! return if workflow.completed? || workflow.halted? perform_next_step_later end private def check_provider_status = # ... end
44. PIPELINES: STEP MACHINE class Cable::DiagnosticJob < ApplicationJob include ActiveJob::Continuable attribute :metadata, default: {} attribute :error_msg, :string def perform(cable) @diagnostic = Cable::Diagnostic.new(cable) step step step step end :provider_status, isolated: true :websocket_status, isolated: true unless error_msg :admin_api_status, isolated: true unless error_msg :save_metadata private def provider_status # ... end = check(:provider_status)
45. PIPELINES: STEP MACHINE class Cable::DiagnosticJob < ApplicationJob include ActiveJob::Continuable attribute :metadata, default: {} attribute :error_msg, :string def perform(cable) = # ... private def check(name) result = diagnostic.perform_check(name) metadata[name] = result.data if result.level == :success broadcast_update else self.error_msg = result.reason end end end
46. PIPELINES ON #step What's missing?
47. PIPELINES ON #step What's missing: - observability Step? Which one? Since when?
48. PIPELINES ON #step What's missing: - observability - per-step rules (callbacks / retry / halt) # app/jobs/account/data_import_job.rb @fizzy class Account::DataImportJob < ApplicationJob include ActiveJob::Continuable TERMINAL_ERRORS = [RecordSet::IntegrityError, ...] discard_on(*TERMINAL_ERRORS) def perform(import) step :check step :process end private def resume_job(exception) TERMINAL_ERRORS.any? { exception.is_a?(it) } ? raise(exception) : super end end
49. PIPELINES ON #step What's missing: - observability - per-step rules - uniqueness Account::DataImportJob.perform_later(import)
50. PIPELINES ON #step What's missing: - observability - per-step rules - uniqueness Account::DataImportJob.perform_later(import) Account::DataImportJob.perform_later(import) Account::DataImportJob.perform_later(import)
51. PIPELINES: GEMS fractaledmind/acidic_jobs ↳ durable state, unique_by, discard_on works for steps, Active Job interface class Account::DataImportJob < ApplicationJob include AcidicJob::Workflow TERMINAL_ERRORS = [RecordSet::IntegrityError, ...] discard_on(*TERMINAL_ERRORS) def perform(import) @import = import execute_workflow(unique_by: import.id) do |w| w.step :check w.step :process end end def check = # ... def process = # ... end
52. PIPELINES: GEMS fractaledmind/acidic_jobs julik/geneva_drive ↳ durable/observable state, cancel_on, callbacks class CardGeneration < GenevaDrive::Workflow alias card hero cancel_if { card.failed? } step :prepare step :moderate step :generate def prepare = # ... # ... def after_step_execution(_) = Turbo::StreamsChannel.broadcast_refresh_to(card) end
53. PIPELINES: GEMS fractaledmind/acidic_jobs julik/geneva_drive ↳ durable/observable state, cancel_on, callbacks, but... no cursors, custom DSL class CardGeneration < GenevaDrive::Workflow # ... end # trigger CardGeneration.create!(hero: card) # observe workflow = CardGeneration.for_hero(card).ongoing.first workflow.current_step_name #=> "moderate" workflow.execution_history.each do puts "#{it.step_name}: #{it.state}" end
54. PATTERNS Pipelines Batches
55. BATCHES Serial batches are just pipelines # app/jobs/imports/process_job.rb @spree class Spree::Imports::ProcessJob < ApplicationJob include ActiveJob::Continuable def perform(import_id) @import = Spree::Import.find(import_id) step :begin_processing step :create_rows, start: 1 # fan-out return if @csv_failed step :reset_row_counters # fan-in end private def create_rows(step) CSV.foreach(...) do |csv_row| next if row_number < step.cursor # ... step.set!(row_number) end end end
56. BATCHES Serial batches are just pipelines Parallel tasks require workarounds (counters, polling) # app/workers/import/... @gitlab class Gitlab::GithubImport::ImportPullRequestsWorker include ApplicationWorker def import(client, project) waiter = PullRequestsImporter.new( project, client).execute AdvanceStageWorker.perform_async(project.id, {waiter.key => waiter.jobs_remaining}) end end class Gitlab::Import::AdvanceStageWorker def perform(project_id, waiters) new_waiters = wait_for_jobs(waiters) if new_waiters.empty? proceed_to_next_stage(next_stage, project_id) else self.class.perform_in(INTERVAL, project_id, new_waiters) # polling end end end
57. BATCHES Serial batches are just pipelines Parallel tasks require workarounds (counters, polling) ↳ Third-party gems are popular # app/workers/import/... @gitlab class Gitlab::GithubImport::ImportPullRequestsWorker include ApplicationWorker def import(client, project) waiter = PullRequestsImporter.new( project, client).execute AdvanceStageWorker.perform_async(project.id, {waiter.key => waiter.jobs_remaining}) end end class Gitlab::Import::AdvanceStageWorker def perform(project_id, waiters) new_waiters = wait_for_jobs(waiters) if new_waiters.empty? proceed_to_next_stage(next_stage, project_id) else self.class.perform_in(INTERVAL, project_id, new_waiters) # polling end end end
58. SOLID QUEUE 1.7 jpcamara.com SolidQueue::Batch !!!
59. SOLID QUEUE 1.7 jpcamara.com SolidQueue::Batch 2.5 years in works Promise-driven development
60. SOLID QUEUE 1.7 SolidQueue::Batch 2.5 years in works Everything is a job SolidQueue::Batch.enqueue( on_finish: OnFinishJob, on_success: OnSuccessJob, on_failure: OnFailureJob, description: "It's never too many jobs", metadata: { user_id: } ) do MyJob.perform_later MyOtherJob.perform_later end
61. BATCHES: GEMS Solid Queue 1.7+ GoodJob Sidekiq Pro doximity/simplekiq julik/scatter_gather
62. BATCHES: GEMS Solid Queue 1.7+ GoodJob Sidekiq Pro doximity/simplekiq julik/scatter_gather class GithubImportOrchestrationJob include Simplekiq::OrchestrationJob def perform_orchestration(project_id) @project = Project.find(project_id) client = @project.github_client run ImportRepositoryJob, project_id in_parallel do client.pull_requests(@project).each { run ImportPullRequestJob, project_id, it.number } end run FinishImportJob, project_id end def on_death(status, options) = # ... end
63. BATCHES ON #step What's missing? Honey, should we #step here?
64. BATCHES ON #step What's missing: - batch interface in Active Job
65. PATTERNS Pipelines Batches Timers
66. TIMERS: SCHEDULE Cron-driven invocation # config/solid_queue.yml @anycable-plus license_expiration_reminder: { class: License::ReminderJob, args: [60], schedule: "26 * * * *" } license_expiration: { class: License::ExpirationJob, schedule: "13 */2 * * *" } license_revoke: { class: License::RevokeAccessJob, schedule: "51 */12 * * *" }
67. TIMERS: SCHEDULE Cron-driven invocation A sweep job and many task jobs # app/jobs/license/reminder_job.rb @anycable-plus class License::ReminderJob < ApplicationJob def perform(interval, now = Time.current) License.where(expires_at: range).find_each do LicenseDelivery .license_expiring_two_weeks(it).deliver_later end end end class License::RevokeAccessJob < ApplicationJob def perform(now = Time.current) License.expired.where(expires_at: ...2.weeks.ago) .find_each(&:revoke_later) end end class License::ExpirationJob < ApplicationJob def perform = # ... end
68. TIMERS: SCHEDULE Cron-driven invocation A sweep job and many task jobs Repeat for every workflow # app/jobs/license/reminder_job.rb @anycable-plus class License::ReminderJob < ApplicationJob def perform(interval, now = Time.current) License.where(expires_at: range).find_each do LicenseDelivery .license_expiring_two_weeks(it).deliver_later end end end class License::RevokeAccessJob < ApplicationJob def perform(now = Time.current) License.expired.where(expires_at: ...2.weeks.ago) .find_each(&:revoke_later) end end class License::ExpirationJob < ApplicationJob def perform = # ... end
69. TIMERS: SCHEDULE Cron-driven invocation A sweep job and many task jobs / steps Repeat for every workflow # config/recurring.yml @fizzy production: incineration: class: "Account::IncinerateDueJob" schedule: every 8 hours at minute 16 # app/jobs/account/incinerate_due_job.rb class Account::IncinerateDueJob < ApplicationJob include ActiveJob::Continuable def perform step :incineration do |step| Account.due_for_incineration.find_each { it.incinerate; step.checkpoint! } end end end
70. TIMERS: WAIT A job parked in the future class AppointmentNotificationJob < ApplicationJob def self.schedule(appt) set(wait_until: appt.start_at - 30.minutes) .perform_later(appt, appt.updated_at.to_i) end def perform(appt, ts) appt.with_lock do next unless appt.updated_at.to_i == ts AppointmentNotification.with(appointment:) .deliver_later(appointment.patient) end end end
71. TIMERS: WAIT A job parked in the future Invalidation? Too many jobs waiting? class AppointmentNotificationJob < ApplicationJob def self.schedule(appt) set(wait_until: appt.start_at - 30.minutes) .perform_later(appt, appt.updated_at.to_i) end def perform(appt, ts) # invalidation timestamp appt.with_lock do next unless appt.updated_at.to_i == ts AppointmentNotification.with(appointment:) .deliver_later(appointment.patient) end end end
72. TIMERS: GEMS radioactive-labs/ chrono_forge class Account::IncinerationWorkflow < ApplicationJob prepend ChronoForge::Executor def perform(account_id:) @account = Account.find(account_id) wait 30.days, :grace_period durably_execute :incinerate end def incinerate = @account.incinerate end
73. TIMERS: GEMS radioactive-labs/ chrono_forge julik/geneva_drive class License::Lifecycle < GenevaDrive::Workflow cancel_if { hero.revoked? } step :remind, wait: 50.weeks step :expire, wait: 2.weeks step :revoke, wait: 2.weeks def remind = # ... def expire = # ... def revoke = # ... end # cancel and restart on renewal License::Lifecycle.for_hero(license) .ongoing.each(&:cancel!)
74. TIMERS: SERVICES Temporal ↳ Durable timers, signal instead of cancel-restart class AppointmentReminderWorkflow < Temporalio::Workflow::Definition workflow_init def initialize(appointment_id, start_at) @timer = UpdatableTimer.new(Time.at(Rational(start_at)) - 30 * 60) end workflow_signal def rescheduled(start_at) @timer.wake_up_time = Time.at(Rational(start_at)) - 30 * 60 end def execute(appointment_id, _start_at) @timer.sleep Temporalio::Workflow.execute_activity( SendAppointmentReminder, appointment_id, schedule_to_close_timeout: 60 ) end end
75. PATTERNS Pipelines Batches Timers Signals, or approval flows
76. SIGNALS: AD-HOC Flow split: before and after the signal # app/controllers/.../imports_controller.rb @mastodon class Settings::ImportsController def create @import = current_account .bulk_imports.create!(import_params) BulkImportParseWorker.perform_async(@import.id) redirect_to settings_import_path(import) end def confirm # signal @import.update!(state: :scheduled) BulkImportWorker.perform_async(@import.id) redirect_to settings_imports_path end end
77. SIGNALS: AD-HOC Flow split: before and after the signal Stalled flows garbage collection # app/controllers/.../imports_controller.rb @mastodon class Settings::ImportsController def create @import = current_account class Vacuum::ImportsVacuum .bulk_imports.create!(import_params) def perform BulkImportParseWorker.perform_async(@import.id) BulkImport.confirmation_missed redirect_to settings_import_path(import) end .in_batches .delete_all def confirm # signal end @import.update!(state: :scheduled) end BulkImportWorker.perform_async(@import.id) redirect_to settings_imports_path end end
78. SIGNALS: AD-HOC Flow split: before and after the signal Stalled flows garbage collection Buttons, webhooks, emails, etc. class WithdrawFundsJob < ApplicationJob def perform(payout, wallet_id, amount) Wallet::Withdraw.(wallet_id:, amount:, payout:) end end class PaymentWebhooksController def create # incoming webhook ProcessPaymentJob.perform_later( *params.slice(:token, :status, :amount) ) head :ok end end class ProcessPaymentJob < ApplicationJob def perform(token, status, amount) payout = Payout.find_by!(token:) Ledger.credit!(payout.account, amount) payout.completed! end end
79. SIGNALS: GEMS radioactive-labs/ chrono_forge ↳ continue_if pauses, perform_later resumes class PayoutWorkflow < ApplicationJob prepend ChronoForge::Executor def perform(payout_id:, wallet_id:, amount:) @payout = Payout.find(payout_id) @wallet_id, @amount = wallet_id, amount durably_execute :withdraw continue_if :payment_received?, name: "payment" durably_execute :credit end # ... end class PaymentWebhooksController def create payout = Payout.find_by!(token: params[:token]) payout.update!(payment_params) PayoutWorkflow.perform_later( "payout-#{payout.id}", payout_id: payout.id) head :ok end end
80. SIGNALS: SERVICES Tempo... class PayoutWorkflow < Temporalio::Workflow::Definition attr_accessor :status, :payment workflow_signal def payment_received(token, status, amount) self.payment = {token:, status:, amount:} end def execute(payout_id, wallet_id, amount) self.status = "withdrawing" Temporalio::Workflow.execute_activity(WithdrawFunds, payout_id, wallet_id, amount, start_to_close_timeout: 60) self.status = "waiting_for_payment" Temporalio::Workflow.wait_condition { payment.present? } raise Temporalio::Error::ApplicationError.new("payment failed", non_retryable: true) if payment[:status] == "failed" self.status = "crediting" Temporalio::Workflow.execute_activity(CreditLedger, payout_id, payment[:amount], start_to_close_timeout: 60) self.status = "completed" end end
81. SIGNALS: SERVICES Temporal class PayoutWorkflow < Temporalio::Workflow::Definition attr_accessor :status, :payment workflow_signal def payment_received(token, status, amount) self.payment = {token:, status:, amount:} end def execute(payout_id, wallet_id, amount) self.status = "withdrawing" Temporalio::Workflow.execute_activity(WithdrawFunds, payout_id, wallet_id, amount, start_to_close_timeout: 60) self.status = "waiting_for_payment" Temporalio::Workflow.wait_condition { payment.present? } raise Temporalio::Error::ApplicationError.new("payment failed", non_retryable: true) if payment[:status] == "failed" self.status = "crediting" Temporalio::Workflow.execute_activity(CreditLedger, payout_id, payment[:amount], start_to_close_timeout: 60) self.status = "completed" end end class WithdrawFunds < Temporalio::Activity::Definition def execute(payout_id, wallet_id, amount) Wallet::Withdraw.(wallet_id:, amount:, payout: Payout.find(payout_id)) end end class CreditLedger < Temporalio::Activity::Definition def execute(payout_id, amount) payout = Payout.find(payout_id) Ledger.credit!(payout.account, amount) payout.completed! end end
82. SIGNALS: SERVICES class PayoutWorkflow < Temporalio::Workflow::Definition attr_accessor :status, :payment Temporal workflow_signal def payment_received(token, status, amount) self.payment = {token:, status:, amount:} end Damn, you write Ruby like that?! def execute(payout_id, wallet_id, amount) self.status = "withdrawing" Temporalio::Workflow.execute_activity(WithdrawFunds, payout_id, wallet_id, amount, start_to_close_timeout: 60) self.status = "waiting_for_payment" Temporalio::Workflow.wait_condition { payment.present? } raise Temporalio::Error::ApplicationError.new("payment failed", non_retryable: true) if payment[:status] == "failed" self.status = "crediting" Temporalio::Workflow.execute_activity(CreditLedger, payout_id, payment[:amount], start_to_close_timeout: 60) self.status = "completed" end end class WithdrawFunds < Temporalio::Activity::Definition def execute(payout_id, wallet_id, amount) Wallet::Withdraw.(wallet_id:, amount:, payout: Payout.find(payout_id)) end end class CreditLedger < Temporalio::Activity::Definition def execute(payout_id, amount) payout = Payout.find(payout_id) Ledger.credit!(payout.account, amount) payout.completed! end end
83. PATTERNS Pipelines Batches Timers Signals
84. PATTERNS TOOLS Pipelines No tool: ad-hoc job juggling Batches Timers Signals Third-party gems geneva_drive, acidic_jobs, ChronoForge, simplekiq, gush External services Temporal, hatchet.run, Ductwork
85. PATTERNS TOOLS Pipelines No tool: ad-hoc job juggling Batches Timers Signals Batteries included? These are sold separately! Third-party gems geneva_drive, acidic_jobs, ChronoForge, simplekiq, gush External services Temporal, hatchet.run, Ductwork
87. LET'S BUILD WHAT'S MISSING ON TOP OF THE FOUNDATION
88. INTRODUCING DURABLE CONTINUATIONS
89. DURABLE Survives SIGKILLS, not only SIGTERMS Every step and cursor move is committed A job is a record, not just a payload
90. DURABLE CONTINUATION: THE PLAN State: recorded, identified, observable Pause and resume: on a timer or a signal Step-level lifecycle hooks ↳ An Active Job extension, not a replacement
91. DURABLE STATE class CardGenerationJob < ApplicationJob include ActiveJob::Continuable attribute :verdict def perform(card) @card = card step :prepare step :moderate, isolated: true unless card.failed? step :generate, isolated: true unless card.failed? end end
92. DURABLE STATE class CardGenerationJob < ApplicationJob include ActiveJob::Durable active_job_durable_runs attribute :verdict def perform(card) @card = card job_class step :prepare step :moderate, isolated: true unless card.failed? step :generate, isolated: true unless card.failed? end end key status step state
93. DURABLE STATE class CardGenerationJob < ApplicationJob include ActiveJob::Durable active_job_durable_runs attribute :verdict def perform(card) @card = card job_class CardGeneratorJob step :prepare step :moderate, isolated: true unless card.failed? step :generate, isolated: true unless card.failed? end end CardGenerationJob.perform_later(card) key status card/42 enqueued step state
94. DURABLE STATE class CardGenerationJob < ApplicationJob include ActiveJob::Durable active_job_durable_runs attribute :verdict def perform(card) @card = card job_class key status CardGeneratorJob card/42 running step :prepare step :moderate, isolated: true unless card.failed? step :generate, isolated: true unless card.failed? end end CardGenerationJob.perform_later(card) step state
95. DURABLE STATE class CardGenerationJob < ApplicationJob include ActiveJob::Durable active_job_durable_runs attribute :verdict def perform(card) @card = card job_class key status step CardGeneratorJob card/42 running prepare step :prepare step :moderate, isolated: true unless card.failed? step :generate, isolated: true unless card.failed? end end CardGenerationJob.perform_later(card) state
96. DURABLE STATE class CardGenerationJob < ApplicationJob include ActiveJob::Durable active_job_durable_runs attribute :verdict def perform(card) @card = card job_class key status ... card/42 running step :prepare step :moderate, isolated: true unless card.failed? step :generate, isolated: true unless card.failed? end end CardGenerationJob.perform_later(card) step state moderate {verdict: "unsure"}
97. DURABLE STATE class CardGenerationJob < ApplicationJob include ActiveJob::Durable active_job_durable_runs attribute :verdict def perform(card) @card = card job_class key status ... card/42 running step :prepare step :moderate, isolated: true unless card.failed? step :generate, isolated: true unless card.failed? end end CardGenerationJob.perform_later(card) step state generate {verdict: "unsure"}
98. DURABLE STATE class CardGenerationJob < ApplicationJob include ActiveJob::Durable active_job_durable_runs attribute :verdict def perform(card) @card = card job_class key status ... card/42 running step :prepare step :moderate, isolated: true unless card.failed? step :generate, isolated: true unless card.failed? end end CardGenerationJob.perform_later(card) CardGenerationJob.workflow_runs.for(card).last #=> #<ActiveJob::Durable::Run status: "running", step: "generate", state: {"verdict" => "unsure"}> step state generate {verdict: "unsure"}
99. DURABLE STATE class Event::WebhookDispatchJob < ApplicationJob include ActiveJob::Continuable def perform(event) step :dispatch do |step| Webhook.active.triggered_by(event) .find_each(start: step.cursor) do |webhook| webhook.trigger(event) step.advance! from: webhook.id end end end end
100. DURABLE STATE class Event::WebhookDispatchJob < ApplicationJob include ActiveJob::Continuable def perform(event) step :dispatch do |step| Webhook.active.triggered_by(event) [Webhooks] .find_each(start: step.cursor) do |webhook| [Webhooks] webhook.trigger(event) step.advance! from: webhook.id [Webhooks] end end Killed: 9 end end Trigger #42 Trigger #43 Trigger #44
101. DURABLE STATE class Event::WebhookDispatchJob < ApplicationJob include ActiveJob::Continuable def perform(event) step :dispatch do |step| Webhook.active.triggered_by(event) [Webhooks] .find_each(start: step.cursor) do |webhook| [Webhooks] webhook.trigger(event) step.advance! from: webhook.id [Webhooks] end end Killed: 9 end end Trigger #42 Trigger #43 Trigger #44 [Webhooks] Trigger #42 [Webhooks] Trigger #43 [Webhooks] Trigger #44
102. DURABLE STATE class Event::WebhookDispatchJob < ApplicationJob include ActiveJob::Durable def perform(event) step :dispatch do |step| Webhook.active.triggered_by(event) .find_each(start: step.cursor) do |webhook| webhook.trigger(event) step.advance! from: webhook.id end end end end
103. DURABLE STATE class Event::WebhookDispatchJob < ApplicationJob include ActiveJob::Durable def perform(event) step :dispatch do |step| Webhook.active.triggered_by(event) [Webhooks] Trigger #42 .find_each(start: step.cursor) do |webhook| UPDATE active_job_durable_steps webhook.trigger(event) step.advance! from: webhook.id [Webhooks] Trigger #43 end UPDATE active_job_durable_steps end [Webhooks] Trigger #44 end end SET cursor = ... SET cursor = ... Killed: 9 [Webhooks] Trigger #44 UPDATE active_job_durable_steps SET cursor = ...
104. LIFECYCLE class Cable::DiagnosticJob < ApplicationJob include ActiveJob::Continuable def perform(cable) @diagnostic = Cable::Diagnostic.new(cable) step :provider_status, isolated: true unless error_msg step :websocket_status, isolated: true step :admin_api_status, isolated: true end step :save_metadata end def check(name) result = diagnostic.perform_check(name) metadata[name] = result.data if result.level == :success broadcast_update else self.error_msg = result.reason end end end
105. LIFECYCLE class Cable::DiagnosticJob < ApplicationJob include ActiveJob::Continuable def perform(cable) @diagnostic = Cable::Diagnostic.new(cable) step :provider_status, isolated: true unless error_msg step :websocket_status, isolated: true step :admin_api_status, isolated: true end step :save_metadata end def check(name) result = diagnostic.perform_check(name) metadata[name] = result.data if result.level == :success broadcast_update else self.error_msg = result.reason end end end class Cable::DiagnosticJob < ApplicationJob include ActiveJob::Durable after_step :broadcast_update def perform(cable) @diagnostic = Cable::Diagnostic.new(cable) step :provider_status, isolated: true step :websocket_status, isolated: true step :admin_api_status, isolated: true step :save_metadata end def check(name) result = diagnostic.perform_check(name) unless result.level == :success halt!(result.reason) end end
106. LIFECYCLE class Account::DataImportJob < ApplicationJob include ActiveJob::Continuable TERMINAL_ERRORS = [RecordSet::IntegrityError, ...] discard_on(*TERMINAL_ERRORS) def perform(import) @import = import step :check step :process end private def resume_job(exception) TERMINAL_ERRORS.any? { exception.is_a?(it) } ? raise(exception) : super end end Account::DataImportJob.perform_later(import) Account::DataImportJob.perform_later(import) Account::DataImportJob.perform_later(import)
107. LIFECYCLE class Account::DataImportJob < ApplicationJob include ActiveJob::Continuable class Account::DataImportJob < ApplicationJob include ActiveJob::Durable TERMINAL_ERRORS = [RecordSet::IntegrityError, ...] discard_on(*TERMINAL_ERRORS) discard_on RecordSet::IntegrityError, ZipFile::InvalidFileError halt_on InsufficientStorageSpaceError def perform(import) @import = import unique_by :import, on_conflict: :skip # or reject, or replace step :check step :process end def perform(import) @import = import step :check private def resume_job(exception) step :process TERMINAL_ERRORS.any? { exception.is_a?(it) } ? end raise(exception) : end super end end Account::DataImportJob.perform_later(import) Account::DataImportJob.perform_later(import) # ignored Account::DataImportJob.perform_later(import) Account::DataImportJob.perform_later(import) # ignored Account::DataImportJob.perform_later(import) Account::DataImportJob.perform_later(import)
108. TIMES AND WAITS class License::ReminderJob < ApplicationJob def perform(interval, now = Time.current) License.where(expires_at: range).find_each do LicenseDelivery .license_expiring_two_weeks(it).deliver_later end end end class License::RevokeAccessJob < ApplicationJob def perform(now = Time.current) License.expired.where(expires_at: ...2.weeks.ago) .find_each(&:revoke_later) end end class License::ExpirationJob < ApplicationJob def perform = # ... end
109. TIMES AND WAITS class License::ReminderJob < ApplicationJob class License::LifecycleJob < ApplicationJob def perform(interval, now = Time.current) include ActiveJob::Durable License.where(expires_at: range).find_each do LicenseDelivery unique_by :license, on_conflict: :replace .license_expiring_two_weeks(it).deliver_later end def perform(license) end @license = license end step :remind, class License::RevokeAccessJob < ApplicationJob wait_until: license.expires_at - 2.weeks def perform(now = Time.current) step :expire, wait_until: license.expires_at License.expired.where(expires_at: ...2.weeks.ago) step :revoke, wait: 2.weeks .find_each(&:revoke_later) end end end end class License::ExpirationJob < ApplicationJob def perform = # ... end # on renewal: cancel - restart (see .unique_by) License::LifecycleJob.perform_later(license)
110. TIMES AND WAITS # config/recurring.yml production: incineration: class: "Account::IncinerateDueJob" schedule: every 8 hours at minute 16 # app/jobs/account/incinerate_due_job.rb class Account::IncinerateDueJob < ApplicationJob include ActiveJob::Continuable def perform step :incineration do |step| Account.due_for_incineration.find_each { it.incinerate; step.checkpoint! } end end end
111. TIMES AND WAITS # config/recurring.yml production: incineration: class: "Account::IncinerateDueJob" schedule: every 8 hours at minute 16 class Account::IncinerationJob < ApplicationJob include ActiveJob::Durable unique_by :account def perform(account) # app/jobs/account/incinerate_due_job.rb step :incinerate, wait: 30.days do class Account::IncinerateDueJob < ApplicationJob account.incinerate include ActiveJob::Continuable end end def perform end step :incineration do |step| Account.due_for_incineration.find_each { class Account::Cancellation < ApplicationRecord it.incinerate; step.checkpoint! # ... } has_one_performed :incineration_job, end dependent: :cancel end # IncinerationJob.workflow_runs.for(account).live.first end end
112. TIMES AND WAITS Stored: runs.wake_at Triggered: one recurrent job for all alarms (and whatever scheduler you use)
113. TIMES AND WAITS Stored: runs.wake_at Triggered: one recurrent job for all alarms (and whatever scheduler you use) ↳ Nothing waits in the queue Does he know the talk is only 30 minutes?
114. SIGNALS class PayoutWorkflow < Temporalio::Workflow::Definition attr_accessor :status, :payment workflow_signal def payment_received(token, status, amount) self.payment = {token:, status:, amount:} end def execute(payout_id, wallet_id, amount) self.status = "withdrawing" Temporalio::Workflow.execute_activity(WithdrawFunds, payout_id, wallet_id, amount, start_to_close_timeout: 60) self.status = "waiting_for_payment" Temporalio::Workflow.wait_condition { payment.present? } raise Temporalio::Error::ApplicationError.new("payment failed", non_retryable: true) if payment[:status] == "failed" self.status = "crediting" Temporalio::Workflow.execute_activity(CreditLedger, payout_id, payment[:amount], start_to_close_timeout: 60) self.status = "completed" end end
115. SIGNALS class PayoutWorkflow < Temporalio::Workflow::Definition attr_accessor :status, :payment workflow_signal def payment_received(token, status, amount) self.payment = {token:, status:, amount:} end class PayoutJob < ApplicationJob include ActiveJob::Durable unique_by :payout, on_conflict: :reject attribute :payment def perform(payout, def execute(payout_id, wallet_id, amount) self.status = "withdrawing" @payout = payout Temporalio::Workflow.execute_activity(WithdrawFunds, payout_id, step :withdraw wallet_id, amount, start_to_close_timeout: 60) wallet_id, amount) await :payment, wait: 3.days step :credit self.status = "waiting_for_payment" Temporalio::Workflow.wait_condition { payment.present? } end raise Temporalio::Error::ApplicationError.new("payment failed", non_retryable: true) if payment[:status] == "failed" def payment(res) halt!(:payment_overdue) if res.nil? self.status = "crediting" Temporalio::Workflow.execute_activity(CreditLedger, payout_id, halt!(:payment_failed) if res.status == "failed" payment[:amount], start_to_close_timeout: 60) self.payment = res.amount self.status = "completed" end end end end
116. SIGNALS # card_generation_job.rb @sfruby-clouds-2026 class CardGenerationJob < ApplicationJob include ActiveJob::Continuable def perform(card) @card = card step :prepare, isolated: true step :moderate, isolated: true unless card.analyzed? || card.needs_review? return if card.needs_review? step :generate, isolated: true unless card.failed? end end # on admin approval (human-in-the-loop) card.analyzed!; CardGenerationJob.perform_later(card)
117. SIGNALS # card_generation_job.rb @sfruby-clouds-2026 class CardGenerationJob < ApplicationJob include ActiveJob::Continuable def perform(card) @card = card step :prepare, isolated: true step :moderate, isolated: true unless card.analyzed? || card.needs_review? return if card.needs_review? step :generate, isolated: true unless card.failed? end end # on admin approval (human-in-the-loop) card.analyzed!; CardGenerationJob.perform_later(card) Durability lives in card.state Idempotency brings resumability
118. SIGNALS # card_generation_job.rb @sfruby-clouds-2026 class CardGenerationJob < ApplicationJob include ActiveJob::Continuable def perform(card) @card = card step :prepare, isolated: true step :moderate, isolated: true unless card.analyzed? || card.needs_review? return if card.needs_review? step :generate, isolated: true unless card.failed? end end # on admin approval (human-in-the-loop) card.analyzed!; CardGenerationJob.perform_later(card) Durability lives in card.state Idempotency brings resumability Races? Execution history? Deadlines?
119. SIGNALS class CardGenerationJob < ApplicationJob # card_generation_job.rb @sfruby-clouds-2026 include ActiveJob::Durable class CardGenerationJob < ApplicationJob include ActiveJob::Continuable attribute :verdict, :string def perform(card) def perform(card) @card = card @card = card step :prepare, isolated: true step :prepare, isolated: true step :moderate, isolated: true unless step :moderate, isolated: true card.analyzed? || card.needs_review? await :review, wait: 1.hour if needs_review? return if card.needs_review? step :generate, isolated: true unless rejected? step :generate, isolated: true unless card.failed? end end end def review(val) = self.verdict = val || "rejected" end # on admin approval (human-in-the-loop) card.analyzed!; CardGenerationJob.perform_later(card) # on admin approval (human-in-the-loop) card.generation_run.wake_up(:review, "approved")
120. SIGNALS class CardGenerationJob < ApplicationJob include ActiveJob::Durable attribute :verdict, :string def perform(card) @card = card step :prepare, isolated: true step :moderate, isolated: true await :review, wait: 1.hour if needs_review? step :generate, isolated: true unless rejected? end def review(val) = self.verdict = val || "rejected" end # on admin approval (human-in-the-loop) card.generation_run.wake_up(:review, "approved")
121. SIGNALS await is a step with an empty body
122. SIGNALS await is a step with an empty body wake_up is a state update from the outside
123. SIGNALS await is a step with an empty body wake_up is a state update from the outside ↳ Signals are timers you can wake up early
124. BONUS: ruby_llm 2.0 class AgentRunJob < ApplicationJob include ActiveJob::Durable def perform(chat) step :run do |step| until chat.complete? chat.step if chat.awaiting_approval? halt!(:tool_approval) end step.checkpoint! end end end end class ApprovalsController < ApplicationController def create chat = Chat.find(params[:chat_id]) params[:approved] == "true" ? chat.approve(params[:tool_call_id]) : chat.deny(params[:tool_call_id]) chat.agent_run_job.resume! end end
125. DURABLE CONTINUATION: THE PLAN State: recorded, identified, observable Pause and resume: on a timer or a signal Step-level lifecycle hooks
126. DURABLE CONTINUATION: THE GEM State: recorded, identified, observable Pause and resume: on a timer or a signal Step-level lifecycle hooks ↳ gem "ajdc"
127. palkan/ github.com/palkan/ajdc Durable workflows on the Continuation foundation Timers, signals, observable state, step-level hooks, AI skills (rails-hyperdrive)
128. palkan/ github.com/palkan/ajdc Durable workflows on the Continuation foundation Timers, signals, observable state, step-level hooks, AI skills (rails-hyperdrive) ↳ Just enough orchestration to stay on the Rails Way
129. palkan/ github.com/palkan/ajdc Durable workflows on the Continuation foundation Timers, signals, observable state, step-level hooks, AI skills (rails-hyperdrive) ↳ Just enough orchestration to stay on the Rails Way #NoPaaS!
130. DURABLE CONTINUATION: MORE PLANS Per-step retry configuration isolate_each and isolate_after Parallel steps / batches Web UI Non-SQL stores
131. DURABLE CONTINUATION: YOUR PLANS Adopt Continuation beyond graceful restarts Identify implicit pipelines, timers, and approval flows in your application Adopt durable workflows
132. THANK YOU Project: github.com/palkan/ajdc Slides: evilmartians.com/events X/Twitter: @palkan_tula, @evilmartians Credits: Ben Orlin / Gemini (Bad Drawings)