Sequences

Added in River Pro for Go v0.5.0.

Sequences are enabled by implementing an optional SequenceOpts() interface on your JobArgs struct:

type MyJobArgs struct {
    CustomerID string `json:"customer_id"`
}

func (MyJobArgs) Kind() string { return "my_job" }

func (MyJobArgs) SequenceOpts() riverpro.SequenceOpts {
    // Use the default sequence partitioning based solely on the job kind.
    return riverpro.SequenceOpts{}
}

The riverpro.SequenceOpts struct configures the sequence partitioning for the job. By default, all jobs of the same kind will run in a sequence.

When a job is inserted, the sequence key is computed automatically based on the sequence options. Typically the returned SequenceOpts should be the same for all jobs of a given kind; River will automatically compute the sequence key based upon those options.

Implement sequence_opts on your argument class, returning River::Pro::SequenceOpts. Empty options sequence all jobs of the same kind:

MyJobArgs = Struct.new(:customer_id, keyword_init: true) do
  def kind = "my_job"
  def to_json = JSON.generate(to_h)
  def sequence_opts = River::Pro::SequenceOpts.new
end

client.insert(MyJobArgs.new(customer_id: "customer_123"))

Insert through a River::Pro::Client so the sequence key is computed automatically. Return consistent options for all jobs of a kind; River derives the key from those options and each job’s encoded arguments.

Sequence options are configured with the riverpro.SequenceOpts type. Using an empty/default SequenceOpts struct will sequence the jobs based solely on the job kind, so that all jobs of that type will run in a single sequence.

This can be customized using the fields on the struct:

type SequenceOpts struct {
    ByArgs bool
    ByQueue bool
    ContinueOnCancelled bool
    ContinueOnDiscarded bool
    ExcludeKind bool
}

Configure partitioning and failure behavior with River::Pro::SequenceOpts. These are the default values:

River::Pro::SequenceOpts.new(
  by_args: false,
  by_queue: false,
  continue_on_cancelled: false,
  continue_on_discarded: false,
  exclude_kind: false
)

Set by_queue: true to keep each queue’s sequence independent. By default the job kind is included, so each kind has its own sequence.

Many use cases will require using some of the job’s arguments, such as a customer or tenant ID, to allow for more granular sequencing. When enabled, ByArgs utilizes all of the job’s encoded arguments.

It’s also possible to use a subset of the args by indicating on the JobArgs struct which fields should be included in the sequence using struct tags:

type MyJobArgs struct {
    CustomerID string `json:"customer_id" river:"sequence"`
    TraceID string `json:"trace_id"`
}

func (MyJobArgs) Kind() string { return "my_job" }

func (MyJobArgs) SequenceOpts() riverpro.SequenceOpts {
    return riverpro.SequenceOpts{ByArgs: true}
}

A sequence can also span across multiple job kinds by setting ExcludeKind to true. In the above example, if ExcludeKind were set to true, the sequence key would be computed based solely on the customer_id field. If other job types were also sequenced in the same way, they would run in a single global sequence for that particular customer ID.

Use by_args: true to include all encoded arguments, or an array of keys to select only fields that define the sequence. Here, jobs for the same customer run sequentially even when their trace IDs differ:

MyJobArgs = Struct.new(:customer_id, :trace_id, keyword_init: true) do
  def kind = "my_job"
  def to_json = JSON.generate(to_h)

  def sequence_opts
    River::Pro::SequenceOpts.new(by_args: [:customer_id])
  end
end

Strings and symbols select encoded JSON keys. Nested fields use path arrays, such as by_args: [[:customer, :id]].

Set exclude_kind: true to share a sequence across job kinds with the same selected arguments. For this example, all participating kinds would share one sequence per customer.

However, this isn’t always desirable. This behavior can be customized with the ContinueOnCancelled and ContinueOnDiscarded options. When set to true, the sequence will continue to be processed even if a job in the sequence is cancelled or discarded:

riverpro.SequenceOpts{
    ContinueOnCancelled: true,
    ContinueOnDiscarded: true,
}

To continue past cancelled or discarded jobs, set the corresponding options to true:

River::Pro::SequenceOpts.new(
  continue_on_cancelled: true,
  continue_on_discarded: true
)