Job Contracts

Have you ever wanted to prevent a background job from writing to the database or perhaps ensure that it completes within a fixed amount of time?

Contracts allow you to easily enforce guarantees like this.

• Why use Contracts?
• Quick Start
• Contracts
  • Breach of Contract
  • Anatomy of a Contract
  • Defining a Contract
  • Using a Contract
• Worker Formation/Topology
• Advanced Usage
• Sidekiq
• Todo
• License
• Sponsors

• Organize your code for better reuse, consistency, and maintainability
• Refine your telemetry and instrumentation efforts
• Improve job performance via enforced (SLAs/SLOs/SLIs)
• Monitor and manage job queue backpressure
• Improve your worker formation/topology to support high throughput

Imagine you want to ensure a specific job completes within 5 seconds of being enqueued.

class ImportantJob < ApplicationJob
  include JobContracts::Contractable

  queue_as :default
  add_contract JobContracts::DurationContract.new(max: 5.seconds)

  def perform
    # logic...
  end

  # default callback that's invoked if the contract is breached
  def contract_breached!(contract)
    # handle breach...
  end
end

How to handle a breach of contract.

A contract is an agreement that a job should fulfill. Failing to satisfy the contract is considered a breach of contract.

Contracts help you track actual results and compare them to expected outcomes. For example, this project has a default set of contracts that verify the following:

• That a job will execute within a set amount of time
• That a job is only performed on a specific queue
• That a job does not write to the database

A breach of contract is similar to a test failure; however, the breach can be handled in many different ways.

• Log and instrument the breach and continue
• Halt processing of the job and all other contracts and raise an exception
• Move the job to a queue where the contract will not be enforced
• etc...

Mix and match any combination of these options to support your requirements.

Contracts support the following constructor arguments.

• trigger [Symbol] (:before, *:after) - when contract enforcement takes place, before or after perform
• halt [Boolean] (true, *false) - indicates whether or not to stop processing when the contract is breached
• queues [Array<String,Symbol>] - a list of queue names where this contract will be enforced (defaults to the configured queue, or * if the queue has not beeen configured)
• expected [Hash] - a dictionary of contract expectations

Here's a contrived, but simple, example that ensures the first argument passed to perform fits within a specific range of values.

# app/contracts/argument_contract.rb
class ArgumentContract < JobContracts::Contract
  def initialize(range:)
    # enforced on all queues
    super queues: ["*"], expected: {range: range}
  end

  def enforce!(contractable)
    actual[:argument] = contractable.arguments.first
    self.satisfied = expected[:range].cover?(actual[:argument])
    super
  end
end

Here's how to use the ArgumentContract in a job.

# app/jobs/argument_example_job.rb
class ArgumentExampleJob < ApplicationJob
  include JobContracts::Contractable

  queue_as :default
  add_contract ArgumentContract.new(range: (1..10))

  def perform(arg)
    # logic...
  end

  # default callback that's invoked if the contract is breached
  def contract_breached!(contract)
    # handle breach...
  end
end

This job will help ensure that the argument passed to perform is between 1 and 10. It's up to you to determine how to handle a breach of contract.

Thoughtful Rails applications often use specialized worker formations.

A simple formation might be to use two sets of workers. One set dedicated to fast low-latency jobs with plenty of dedicated compute resources (CPUs, processes, threads, etc...), with another set dedicated to slower jobs that uses fewer compute resources.

Say we determine that fast low-latency jobs should not write to the database.

We can use a ReadOnlyContract to enforce this decision. If the contract is breached, we can notify our apm/monitoring service and re-enqueue the job to a slower queue (worker set) where database writes are permitted. This will ensure that our fast low-latency queue doesn't get clogged with slow-running jobs.

Here's an example job implementation that accomplishes this.

class FastJob < ApplicationJob
  include JobContracts::Contractable

  # Configure the queue before adding contracts
  # It will be used as the default enforcement queue for contracts
  queue_as :critical

  # Only enforces on the critical queue
  # This allows us to halt job execution and reenqueue the job to a different queue
  # where the contract will not be enforced
  #
  # NOTE: the arg `queues: [:critical]` is default behavior in this example
  #       we're setting it explicitly here for illustration purposes
  add_contract JobContracts::ReadOnlyContract.new(queues: [:critical])

  def perform
    # logic that shouldn't write to the database,
    # but might accidentally due to complex or opaque internals
  end

  def contract_breached!(contract)
    # log and notify apm/monitoring service

    # re-enqueue to a different queue
    # where the database write will be permitted
    # i.e. where the contract will not be enforced
    enqueue queue: :default
  end
end

Worker formations can be designed in countless ways to handle incredibly sophisticated requirements and operational constraints. The only real limitation is your creativity.

It's possible to override the default callback method that handles contract breaches.

class ImportantJob < ApplicationJob
  include JobContracts::Contractable

  queue_as :default
  on_contract_breach :take_action
  add_contract JobContracts::DurationContract.new(max: 5.seconds)

  def perform
    # logic...
  end

  def take_action(contract)
    # handle breach...
  end
end

class ImportantJob < ApplicationJob
  include JobContracts::Contractable

  queue_as :default
  on_contract_breach -> (contract) { # take action... }

  add_contract JobContracts::DurationContract.new(max: 5.seconds)

  def perform
    # logic...
  end
end

Sidekiq::Jobs are also supported.

class ImportantJob
  include Sidekiq::Job
  include JobContracts::SidekiqContractable

  sidekiq_options queue: :default
  add_contract JobContracts::DurationContract.new(max: 1.second)

  def perform
    # logic...
  end

  # default callback that's invoked if the contract is breached
  def contract_breached!(contract)
    # handle breach...
  end
end

• Sidekiq tests

The gem is available as open-source under the terms of the MIT License.

This project is sponsored by Orbit.love (mission control for your community).

This effort was partly inspired by a presentation at Sin City Ruby from our friends on the platform team at Gusto. Their presentation validated some of my prior solutions aimed at accomplishing similar goals and motivated me to extract that work into a GEM.