Sequences and batching

[dannydeloff]

I have a job that I would like to batch, but also be part of a sequence. In the docs it says these two features should work together. However, when I have my job sequenced and batched by the same field level argument, I've noticed that the sequence takes precedence and doesn't allow the jobs to be batched together. (One job gets worked on at a time).

How should these components work together? Could I have something misconfigured?

[bgentry]

Hi @dannydeloff , these should work well together based on how they work, although I'm not sure how much we've tested the combination directly.

The short version of it is that sequences ensure that at most one job from a given sequence is ever allowed past pending (a non-runnable state where jobs are parked until external action makes them available). The exceptions are cases where you've configured ConfigureOnCancelled / ConfigureOnDiscarded to keep the sequence running even if a single job fails. When a previous job in a sequence completes, the next one is moved to available (or scheduled if it has a future ScheduledAt) where it is available for a worker to fetch it.

The above process is totally separate from fetching where any available job is fetched by the normal priority within that queue ((priority, scheduled_at, id)). Batching involves a leader job being fetched and then fetching additional eligible jobs (with the same worker kind and batch key) until a full batch or timeout. Unless you also have concurrency limits enabled on the same partitioning setup, you can end up with multiple batch leaders running at the same time. Sequence jobs which are not yet available (blocked by another job in the sequence) are not eligible for batching.

Is this a situation you're encountering? Could you provide more detail on what you're seeing and how your setup is configured?

If you'd prefer to provide more info by email we're at team [ at ] riverqueue.com.

[dannydeloff]

I believe that's exactly what I'm encountering, where jobs are blocked from batching because they are all in the same sequence. (Sequence keys are identical and Batch keys are identical).

Essentially I have few different jobs that depend upon on the order, let's say A, B, and C. A must be (and will be) created first, and then sometime in the future we may have some jobs for B and C. C depends on changes made in B to finish, while also depending on A (if it's still running). Many jobs from B can happen in quick succession, which is where I would want to implement batching, but is dependent on A (if it's still running). Since A, B, and C all have the same sequence key, jobs of B cannot be batched together.

Here's a simple example, similar to what I have for my worker's args:

type BatchArgs struct {
	InternalID   string `json:"internal_id" river:"batch,sequence"`
}

func (BatchArgs) SequenceOpts() riverpro.SequenceOpts {
	return riverpro.SequenceOpts{ByArgs: true, ExcludeKind: true}
}

func (BatchArgs) BatchOpts() riverpro.BatchOpts {
	return riverpro.BatchOpts{ByArgs: true}
}

And then I am using concurrency limits to avoid the multiple batch leaders that you described above:

ProQueues: map[string]riverpro.QueueConfig{
	"batch-queue": {
		Concurrency: riverpro.ConcurrencyConfig{
			GlobalLimit: 1,
			Partition:   riverpro.PartitionConfig{ByArgs: []string{"internal_id"}},
		},
		MaxWorkers: 100,
	},
},

[bgentry]

@dannydeloff so if I'm understanding you correctly you were expecting that jobs within the same sequence would be batched and executed together. If so, that's not how this would work because it would violate the core promise of sequences: only one job from a given sequence may run at once.

However if you're wanting to batch multiple jobs of type B that all have different sequence keys and which happen to be available at the same time, this should work just fine. I'm not really sure of a mechanism that would prevent it.

It's hard to be sure without knowing more about your use case, but it almost sounds like a use case for workflows since you want certain jobs to depend on multiple other jobs (I think that's what you're saying?). You could use workflows together with concurrency limits on a specific queue that handles these concurrency-limited jobs.