Concurrency limits

Concurrency limiting allows precise control over how many jobs can be worked at once, with configurable partitioning options. Limits can be configured globally, across all processes, and locally, per process, and can also be partitioned based on job attributes.

Concurrency limits are a feature of River Pro ✨. If you haven’t yet, install River Pro and run the pro migration line.

Basic usage

riverpro migrate-up --database-url "$DATABASE_URL" --line pro

&riverpro.Config{
    Config: river.Config{
        Queues: map[string]river.QueueConfig{
            "default": {MaxWorkers: 100},
        },
        Workers: riverdemo.Workers(dbPool),
    },
    ProQueues: map[string]riverpro.QueueConfig{
        "external_api_provider": {
            Concurrency: riverpro.ConcurrencyConfig{
                GlobalLimit: 20,
                LocalLimit:  10,
            },
            MaxWorkers: 20,
        },
    },
}

bundle exec river migrate-up --database-url "$DATABASE_URL" --line pro

River::Pro::Config.new(
  core: River::Config.new(queues: {default: 100}, workers: workers),
  pro_queues: {
    "external_api_provider" => River::Pro::QueueConfig.new(
      concurrency: River::Pro::ConcurrencyConfig.new(
        global_limit: 20,
        local_limit: 10
      ),
      max_workers: 20
    )
  }
)

Concurrency limits build on River’s normal job prioritization, meaning that jobs will still be fetched in order of priority and scheduled_at. However, any job that would be fetched based on normal prioritization will be skipped over if working it would exceed a concurrency limit.

Global limits

Global limits ensure that no more than the specified number of jobs are worked at once globally across all clients working jobs in that queue (and database / schema). Set the global limit in the queue’s concurrency configuration.

If no partitioning is enabled, the global limit applies to all jobs in the queue. With partitioning, the global limit is enforced independently per partition.

Local limits

Local limits ensure that no more than the specified number of jobs are worked at once locally within a single client. Set the local limit in the queue’s concurrency configuration.

If no partitioning is enabled, the local limit applies to all jobs in the queue. With partitioning, the local limit is enforced independently per partition.

Partitioning

By default, concurrency limits are enforced across all jobs in a queue. With partitioning, limits are enforced separately for different subsets of jobs in the queue.

Partitioning can be based on job arguments, job kind, or both.

Partitioning by arguments

Partitioning by job arguments allows limits to be applied independently based on the argument values. A common use case is to partition by customer or tenant, allowing each to have an independent concurrency limit:

&riverpro.Config{
    ProQueues: map[string]riverpro.QueueConfig{
        "expensive_actions": {
            Concurrency: riverpro.ConcurrencyConfig{
                GlobalLimit: 2,
                Partition:   riverpro.PartitionConfig{
                    ByArgs: []string{"customer_id"},
                },
            },
            MaxWorkers: 10,
        },
    },
}

River::Pro::Config.new(
  core: River::Config.new(workers: workers),
  pro_queues: {
    "expensive_actions" => River::Pro::QueueConfig.new(
      concurrency: River::Pro::ConcurrencyConfig.new(
        global_limit: 2,
        partition: River::Pro::PartitionConfig.new(by_args: ["customer_id"])
      ),
      max_workers: 10
    )
  }
)

Here, a maximum of 2 concurrent jobs per customer run globally, while each client can handle jobs for multiple customers concurrently, up to the queue’s configured worker count.

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

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

Partitioning by kind

Partitioning can also be based on job kind. For example, rather than a single global limit of 10 across all job kinds, the following configuration allows up to 10 jobs concurrently for each job kind:

&riverpro.Config{
    ProQueues: map[string]riverpro.QueueConfig{
        "expensive_actions": {
            Concurrency: riverpro.ConcurrencyConfig{
                GlobalLimit: 10,
                Partition:   riverpro.PartitionConfig{ByKind: true},
            },
            MaxWorkers: 100,
        },
    },
}

River::Pro::Config.new(
  core: River::Config.new(workers: workers),
  pro_queues: {
    "expensive_actions" => River::Pro::QueueConfig.new(
      concurrency: River::Pro::ConcurrencyConfig.new(
        global_limit: 10,
        partition: River::Pro::PartitionConfig.new(by_kind: true)
      ),
      max_workers: 100
    )
  }
)

In this example, no more than 10 jobs of any single kind will be worked concurrently across all clients. However, a single client may work up to 100 jobs at once from a mix of kinds.

Adjusting in the UI

The concurrency limit can be overridden in the UI on the queue page. Limit overrides are persisted until removed, or until the queue has had no active clients for a long time.

An example of this can be viewed in the demo app.

FAQ

How do limits interact?

The local limit takes precedence over the global limit. For example, with a global limit of 20 and a local limit of 5, if 10 jobs are already running globally, the client will not fetch more than 5 additional jobs (per partition, if partitioning is enabled).

Each client also respects the queue’s configured worker count. Without partitioning, the smallest of the worker count, the local limit, and remaining global capacity determines how many jobs will be worked. With partitioning, a client may fill its worker slots across partitions, but never exceeds the local or global limit for a single partition.

Why might limits appear exceeded?

Global concurrency limits are strictly enforced during job fetching. Jobs may appear to exceed limits in the database due to asynchronous completion updates, at least when viewed from the jobs table (or job UI in River UI).

River ensures limits are respected at execution time, despite temporary visibility differences. The queue page in River UI reflects the actual running job counts.

Compatibility with other features

Concurrency limits are compatible with all other River and River Pro features like workflows and sequences.