Limit parallel jobs by args

[pablo-ruth]

Hey there :)

I'm trying to limit the number of jobs running at the same time with the same arguments, but I'm not sure how to do it properly. I have a job defined like this:

type ReconcileClient struct {
	ClientID uuid.UUID
}

func (ReconcileClient) Kind() string { return "reconcile_client" }

This job gets inserted every time a user performs an action on their account, using their ClientID as an argument. I want to make sure there's only one "running" job per ClientID. So I tried:

func (ReconcileClient) InsertOpts() river.InsertOpts {
	return river.InsertOpts{
		UniqueOpts: river.UniqueOpts{
			ByArgs:  true,
			ByState: []rivertype.JobState{rivertype.JobStateAvailable, rivertype.JobStatePending, rivertype.JobStateRetryable, rivertype.JobStateRunning},
		},
	}
}

But I have a lot of jobs in my river_job table, so the insertion is very slow. From what I understand from other discussions, it's because I'm not filtering on the default set of states. However, I can't filter on the default states because I want to allow the insertion of new jobs with arguments matching completed jobs.

Is there a more efficient way to handle this?

Thanks!

[bgentry]

@pablo-ruth based on the ByState list there it looks like you're using the old advisory lock based unique jobs implementation. This has just been deprecated in the latest release in favor of a much faster unique index based implementation. The caveat is that you no longer have total control to customize the list of states, but I believe that should be ok. The default list of states includes your list above plus the scheduled and completed states:

ByState: []rivertype.JobState{rivertype.JobStateAvailable, rivertype.JobStateCompleted, rivertype.JobStatePending, rivertype.JobStateRunning, rivertype.JobStateRetryable, rivertype.JobStateScheduled}

You can customize this to remove completed (assuming you want "duplicate" jobs to be allowed with the same unique attrs, even though you want to prevent them from existing in the other states). The scheduled state cannot be removed though, because then River wouldn't be able to move those jobs to available when it's their turn to run. I think based on what you're describing, this should address your use case?

However I want to point out that this isn't quite the same as a limit on parallel job execution. With the default ByState list for unique jobs, you are guaranteed that no duplicate job can exist in the database at all given the chosen unique opts, so this can be used to ensure a parallel limit of exactly one for a given job type / set of unique opts. But if you were hoping for limits greater than 1, there's currently no solution for that. One option is to implement your own locking mechanism within the job, but this won't prevent jobs from holding up worker slots.

[pablo-ruth]

@bgentry thanks for your quick answer! So I need to update to the latest River version to use the new implementation. But I'm still not sure how to customize my list of states. Is it like what I've done here but just removing scheduled from the list?

func (ReconcileClient) InsertOpts() river.InsertOpts {
	return river.InsertOpts{
		UniqueOpts: river.UniqueOpts{
			ByArgs:  true,
			ByState: []rivertype.JobState{rivertype.JobStateAvailable, rivertype.JobStatePending, rivertype.JobStateRunning, rivertype.JobStateRetryable, rivertype.JobStateScheduled}
		},
	}
}

With the new unique jobs implementation, does not having the default list of states no longer introduce a performance issue?

Thanks

[bgentry]

Yes that sample looks correct. If you don't specify ByState you will get the default list of states in my previous response. However if you customize it there are certain states that are required (pending, scheduled, available, running). So as long as you have those 4 in the list you are free to add/remove the others.

With the new unique jobs implementation, does not having the default list of states no longer introduce a performance issue?

Correct, the latest implementation allows customizing the list as I've described above, and if you stay within those constraints the fast index-based unique path will be used. Even better, you can now still use unique jobs when using InsertMany.

In the current release you can still customize ByState further to remove required states, but it will log a warning and fall back to the slow advisory lock path. Soon that will be removed (#614) and removing required unique states will be an error. But I don't think this is likely to cause much trouble—I spent a lot of time reasoning about it and couldn't think of any concrete use cases where removing those initial states was likely to be required.

Let me know how it goes! ⚡

[pablo-ruth]

Amazing, I'll check and let you know !

[pablo-ruth]

@bgentry I pushed the code with your advice into production, and it's working like a charm now! Thanks again ;)