Inserting jobs from Ruby

River supports inserting jobs from Ruby and have them worked in Go, a feature that may be desirable in performance sensitive cases so that jobs can take advantage of Go's considerably faster runtime speed.

Insertion is supported through Rails' ActiveRecord and Sequel.

Your project's Gemfile should contain the riverqueue gem and a driver like riverqueue-sequel (see drivers):

gem "riverqueue"
gem "riverqueue-activerecord"

Initialize a client with:

ActiveRecord::Base.establish_connection("postgres://...")
client = River::Client.new(River::Driver::ActiveRecord.new)

Define a job and insert it:

class SortArgs
  attr_accessor :strings

  def initialize(strings:)
    self.strings = strings
  end

  def kind = "sort"

  def to_json = JSON.dump({strings: strings})
end

insert_res = client.insert(SimpleArgs.new(strings: ["whale", "tiger", "bear"]))
insert_res.job # inserted job row

Job args should:

• Respond to #kind with a unique string that identifies them in the database, and which a Go worker will recognize.
• Response to #to_json with a JSON serialization that'll be parseable as an object in Go.

They may also respond to #insert_opts with an instance of InsertOpts to define insertion options that'll be used for all jobs of the kind.

Inserts take an insert_opts parameter to customize features of the inserted job:

insert_res = client.insert(
  SimpleArgs.new(strings: ["whale", "tiger", "bear"]),
  insert_opts: River::InsertOpts.new(
    max_attempts: 17,
    priority: 3,
    queue: "my_queue",
    tags: ["custom"]
  )
)

Unique jobs are supported through InsertOpts#unique_opts, and can be made unique by args, period, queue, and state. If a job matching unique properties is found on insert, the insert is skipped and the existing job returned.

insert_res = client.insert(args, insert_opts: River::InsertOpts.new(
  unique_opts: River::UniqueOpts.new(
    by_args: true,
    by_period: 15 * 60,
    by_queue: true,
    by_state: [River::JOB_STATE_AVAILABLE]
  )
)

# contains either a newly inserted job, or an existing one if insertion was skipped
insert_res.job

# true if insertion was skipped
insert_res.unique_skipped_as_duplicated

Unique job insertion takes a Postgres advisory lock to make sure that it's uniqueness check still works even if two conflicting insert operations are occurring in parallel. Postgres advisory locks share a global 64-bit namespace, which is a large enough space that it's unlikely for two advisory locks to ever conflict, but to guarantee that River's advisory locks never interfere with an application's, River can be configured with a 32-bit advisory lock prefix which it will use for all its locks:

client = River::Client.new(mock_driver, advisory_lock_prefix: 123456)

Doing so has the downside of leaving only 32 bits for River's locks (64 bits total - 32-bit prefix), making them somewhat more likely to conflict with each other.

Use #insert_many to bulk insert jobs as a single operation for improved efficiency:

num_inserted = client.insert_many([
  SimpleArgs.new(job_num: 1),
  SimpleArgs.new(job_num: 2)
])

Or with InsertManyParams, which may include insertion options:

num_inserted = client.insert_many([
  River::InsertManyParams.new(
    SimpleArgs.new(job_num: 1),
    insert_opts: River::InsertOpts.new(max_attempts: 5)
  ),
  River::InsertManyParams.new(
    SimpleArgs.new(job_num: 2),
    insert_opts: River:InsertOpts.new(queue: "high_priority")
  )
])

No extra code is needed to insert jobs from inside a transaction. Just make sure that one is open from your ORM of choice, call the normal #insert or #insert_many methods, and insertions will take part in it.

ActiveRecord::Base.transaction do
  client.insert(SimpleArgs.new(strings: ["whale", "tiger", "bear"]))
end

DB.transaction do
  client.insert(SimpleArgs.new(strings: ["whale", "tiger", "bear"]))
end

JobArgsHash can be used to insert with a kind and JSON hash so that it's not necessary to define a class:

insert_res = client.insert(River::JobArgsHash.new("hash_kind", {
    job_num: 1
}))

The gem bundles RBS files containing type annotations for its API to support type checking in Ruby through a tool like Sorbet or Steep.

Use River with Rails' ActiveRecord by putting the riverqueue-activerecord driver in your Gemfile:

gem "riverqueue"
gem "riverqueue-activerecord"

Then initialize driver and client:

ActiveRecord::Base.establish_connection("postgres://...")
client = River::Client.new(River::Driver::ActiveRecord.new)

Use River with Sequel by putting the riverqueue-sequel driver in your Gemfile:

gem "riverqueue"
gem "riverqueue-sequel"

Then initialize driver and client:

DB = Sequel.connect("postgres://...")
client = River::Client.new(River::Driver::Sequel.new(DB))