Really Complex Workflows with Batches

Josh Adams edited this page Apr 15, 2016 · 7 revisions

Sidekiq Pro's Batches feature can handle job workflows of any complexity. This page shows how to implement a complex workflow given to me by one Sidekiq Pro customer.

The workflow looks like this, where jobs are blue circles and purple boxes hold jobs which can execute in parallel. All jobs within a purple box must succeed before the workflow can move "down".

complex workflow

Let's call this workflow the Order workflow. Perhaps it represents the series of steps necessary to ship a customer's order.

First of all, we're going to create an overall Batch to represent the entire workflow and then create a child batch to represent the first step in the workflow. That child batch will use a success callback to schedule step 2 in the workflow:

order = ...
overall = Sidekiq::Batch.new
overall.on(:success, 'FulfillmentCallbacks#shipped', 'oid' => order.id)
overall.description = "Fulfillment for Order #{order.id}"
overall.jobs do
  step1 = Sidekiq::Batch.new
  step1.on(:success, 'FulfillmentCallbacks#step1_done', 'oid' => order.id)
  step1.jobs do
    A.perform_async(order.id)
  end
end

class FulfillmentCallbacks
  def step1_done(status, options)
    oid = options['oid']
    overall = Sidekiq::Batch.new(status.parent_bid)
    overall.jobs do
      step2 = Sidekiq::Batch.new
      step2.on(:success, 'FulfillmentCallbacks#step2_done', 'oid' => oid)
      step2.jobs do
        B.perform_async
        C.perform_async
        D.perform_async
        E.perform_async
        F.perform_async
      end
    end
  end

  def step2_done(status, options)
    oid = options['oid']
    overall = Sidekiq::Batch.new(status.parent_bid)
    overall.jobs do
      step3 = Sidekiq::Batch.new
      step3.on(:success, 'FulfillmentCallbacks#step3_done', 'oid' => oid)
      step3.jobs do
        G.perform_async(oid)
      end
    end
  end

  def step3_done(status, options)
    oid = options['oid']
    overall = Sidekiq::Batch.new(status.parent_bid)
    overall.jobs do
      step4 = Sidekiq::Batch.new
      step4.on(:success, 'FulfillmentCallbacks#step4_done', 'oid' => oid)
      step4.jobs do
        H.perform_async(oid)
        I.perform_async(oid)
      end
    end
  end

  def step4_done(status, options)
    oid = options['oid']
    overall = Sidekiq::Batch.new(status.parent_bid)
    overall.jobs do
      J.perform_async(oid)
      K.perform_async(oid)
      L.perform_async(oid)
    end
  end

  def shipped(status, options)
    # this callback will fire once M has succeeded
    oid = options['oid']
    puts "Order #{oid} has shipped!"
  end
end

class L
  include Sidekiq::Worker

  def perform(oid)
    # do stuff
    if bid
      # if we belong to a batch, assume we're within the fulfillment workflow
      # and need to kick off job M
      batch.jobs do
        M.perform_async(oid)
      end
    end
  end
end

In this manner, we can implement serial steps in the workflow, with the jobs in each step executing in parallel.