[8.x] Add ability to dispatch unique jobs

[paras-malhotra]

Alternative to #35039 (based on @themsaid's feedback) with a different syntax. Both PRs are backwards-compatible with no breaking changes. I'm leaving both open so that one of them could be chosen based on preference. I do prefer this one over the other one.

Differences between both PRs:

1. This PR has a different syntax where you implement a UniqueJob interface, whereas the other PR allows dispatching a unique job by chaining dispatch(...)->unique(...)
2. This PR releases the lock in CallQueuedHandler whereas the other PR releases the lock by registering a "after middleware" on the job.

Motivation:

This PR allows the ability to dispatch unique jobs. So, if a unique job is already dispatched, it will not dispatch another job (with the same key) unless the job has completed processing. See laravel/ideas#2151.

Difference between this PR and WithoutOverlapping:

The difference between WithoutOverlapping and dispatching unique jobs is that the WithoutOverlapping middleware processes jobs uniquely (meaning two jobs with the same key will not be processed in parallel). However, it does not stop jobs (with the same key) from being enqueued if another such job is already in the queue waiting to be processed.

Use Case:

This can be quite useful in scenarios such as search indexing. If a search index job is already enqueued for a resource, you probably don't want another such job to be queued until the previous job has finished processing. Sidekiq Enterprise offers the unique jobs feature as well.

How to Use

If you wish to make your Job a "unique job", just implement the UniqueJob interface like so:

use Illuminate\Contracts\Queue\ShouldBeUnique;

class MyUniqueJob implements ShouldQueue, ShouldBeUnique
{
    /**
     * The number of seconds after which the job will no longer stay unique.
     *
     * @var int
     */
     public $uniqueFor = 3600;

    public function uniqueId()
    {
        return $this->user->id;
    }
}

EDIT: Edited the above syntax to match the final PR that was merged.

[royduin]

I think this option is better because when a job is dispatched from multiple locations the key is at one place; the job itself. But I think the interface is too explicit, checking for the method existence is enough.

But still; I don't think having the lock in the cache is the way to go. Having php artisan cache:clear in a deploy script is a common thing. So after a deploy the lock is gone and jobs end up twice in the queue. Checking the queue for existing jobs is a better approach.

[paras-malhotra]

I think the interface is too explicit, checking for the method existence is enough

This is for backwards-compatibility. Otherwise, we would have to target master as it would be a breaking change - anyone who's implemented the uniqueBy method on his/her job, would automatically promote it to a unique job. Also, I feel the interface seems expressive (like ShouldQueue).

I don't think having the lock in the cache is the way to go.

Unfortunately, there's no way around this. Laravel doesn't treat lock as a first class citizen (e.g. there is no config/lock.php with a driver, connection, etc. like cache). So, there's no real way to say which lock the user wants (array lock, Redis lock, etc.) except if we go through the default cache driver and obtain the lock.

[taylorotwell]

@paras-malhotra Can you describe the behavior in a failure state? If a job fails will it release the lock? And then try to re-gain the lock on its next retry?

From my brief reading of the code it appears that will be the case because of the finally block. Is that our intended behavior. I see Sidekiq's jobs keep the lock until they are retried and execute successfully. I'm not sure what the community would prefer. Personally I'm not sure it matters that much but I'm not sure of Sidekiq's reasoning for that behavior.

[taylorotwell]

I would also be curious to know how this behaves in the context of job batches and job chains.

[taylorotwell]

From my exploration, it looks like if a chain encounters a unique job that can't acquire a lock the chain would just end there. With batches, I think unique jobs would run because they are often bulk pushed onto the queue and don't go through the normal dispatch flow.

I'm not sure what we want the behavior to be. I don't think unique jobs makes much sense in the context of batches but I'm not sure if people have a use case there. I'm honestly not sure if they make sense as part of chains either but I think it's more likely to happen than batches.

As it stands right now we would need to document that unique jobs should not be included in batches or chains.

[paras-malhotra]

@taylorotwell, unique jobs aren't supported in batches or chains for now. Sidekiq doesn't support them either as mentioned in the notes section of the docs on batches.

Currently, how the PR works is that the lock is acquired when dispatching and the lock is released regardless of success/failure/retry. When a job exception occurs, the job is released back to the queue, and since it's not technically "dispatched" it will not acquire a lock again while retrying. I'll need to change the PR to say that the lock is only released if the job has succeeded or "failed" (hit max exceptions or max retries), but will not be released if the job is released back to the queue (as the lock will not again be acquired on retry). Does that behaviour sound good?

[taylorotwell]

Hmm, well it seems strange that on exception the lock would be released and then the job would not attempt to reacquire the lock when it is retried? It seems like this could allow for multiple of a unique job to exist on the queue at the same time?

[paras-malhotra]

@taylorotwell, here are the 3 scenarios:

1. Job success: Simplest of them all, lock acquired on dispatch and released on completion.
2. Job exception/timeout but not failure (doesn't exceed max exceptions or max retries): Job is released back to the queue and lock is not released. When the worker picks the job, it doesnt need to acquire the lock (as the retry is not a dispatch). If it succeeds, the lock is released (per point 1 above). Because the lock is not released when the timeout/exception happened, there is no chance of multiple jobs (new dispatches would be skipped).
3. Job failure (exceeds max exceptions or max retries): Job fails and lock is released so that a new job can be attempted. This job failed and will not be retried so there is no chance of multiple jobs. The first subsequent dispatch of the job will acquire the lock.

So, in all 3 cases, there is no chance of multiple unique jobs to exist on the queue at the same time.

[taylorotwell]

@paras-malhotra

For scenario #2 - how is the lock not released? I see the lock release is the finally of CallQueuedHandler. It will be called if there is an exception during dispatchThroughMiddleware, so I'm confused how the lock will not be released in this scenario?

For scenario #3, where in the code is the lock released when the job fails?

[paras-malhotra]

@taylorotwell I'd change the code accordingly if you're okay with this behaviour. Right now, the code doesnt look like this. I just wanted to confirm if this behaviour is acceptable. If yes, I'll modify the PR accordingly.

[taylorotwell]

Ah OK. Yes, those 3 scenarios sound good as described.

[themsaid]

Instead of having uniqueBy and uniqueFor, I suggest using a uniqueId term for both a class property and method. We've unified the naming for queue configuration keys in laravel 8 so the property and method would have the same name.

Also I think having an interface is not necessary, checking if there's a unique ID set for the job is enough IMO.

[paras-malhotra]

@themsaid I have added uniqueId instead of uniqueBy per your suggestion. uniqueId is the key of the lock (now as a method or property) and uniqueFor is the timeout of the lock.

The UniqueJob interface was for backwards-compatibility (otherwise any job with a uniqueId property or method will automatically promoted to be a unique job) and also felt more explicit (like ShouldQueue). For now I've removed the interface and checked for the method/property existence instead.

@taylorotwell if you would like to go back to the UniqueJob interface, you can simply revert my last commit (
bc45489).

To summarise the behaviour, now there are 4 scenarios:

1. Job success: Lock acquired on dispatch and released on completion.
2. Job released back to the queue: If a job or middleware releases it back to the queue, the lock is not released as it will anyway be retried (without attempting to acquire the lock).
3. Job exception/timeout but not failure (doesn't exceed max exceptions or max retries): Job is released back to the queue and lock is not released as the job will be retried (without attempting to acquire the lock).
4. Job failure (exceeds max exceptions or max retries): Job fails and lock is released so that a new job can be attempted. The first subsequent dispatch of the job will then acquire the lock.

Tests are added for all 4 scenarios. 🚀 Let me know if any further changes are needed.

[taylorotwell]

@themsaid @paras-malhotra there is a use case for the interface IMO. For example, if you mark the class with the interface but do not define any additional methods it will lock by the class name meaning only 1 job of that entire type will be allowed on the queue at once. That is a valid use case IMO.

[themsaid]

@taylorotwell hmmm yes that's a valid use case but wouldn't you want to define a lock timeout in all cases? You'll have to define a $uniqueFor property. we can use that property to determine if the job should be unique or not.

Also if we bring the interface back, I think we can make it optional. So people can use it only if they want that behaviour of using the job class name as the lock key.

[paras-malhotra]

@taylorotwell, I think the interface is useful because of the following:

1. It's automatic like you mentioned. Just implement it if you want the job to be unique (in a global way)
2. It's non-breaking - users don't have to be mindful of overriding the uniqueId method/property (which will auto-promote the job to be unique)

@themsaid I've kept a default value of zero for uniqueFor (meaning it's locked forever until the job has succeeded/failed). We can change that if needed, but simply implementing the interface will make a job unique (no need to define any methods or properties if the interface is implemented).

[taylorotwell]

@paras-malhotra can you add the interface back for now. I don't know how to revert a commit on here.

[paras-malhotra]

@taylorotwell reverted.

[moshe-autoleadstar]

@paras-malhotra @taylorotwell while i love this, and did something similar (albeit much messier 😅 in my own repo), there's an edge case problem with this implementation. job A can start processing, job B checked if should be dispatched and the answer is no, and only then job A finishes. this is especially a problem for a classic case of the "indexing" job (as used in the example of the PR) which can take a while to process. it will index the old data and the newer data will not be queued for reindex.
as a working alternative, consider unlocking right BEFORE handling/processing the job. (as opposed to AFTER the job was processed.)

[lasselehtinen]

I have similar usecase as @moshe-autoleadstar. Did I understood the implementation correctly, if the job with the unique id has already started to process, another job with the same id won't trigger? I am not use are about the queue internals, do the jobs have any status like "queued", "processing" and "complete"? In that case, I would like like to allow the "duplicate" job only if the status is !== "queued".

[paras-malhotra]

I understand the use case but we need to consider what happens on an exception / timeout (causing a retry) or job release? If the lock is released before processing, we may have to re-acquire the lock on a retry/release but that would mean that between the window of releasing the lock and re-acquiring, there may be another job dispatch. Or alternatively, we could simply ignore everything if the job lock is already released.

[moshe-autoleadstar]

@paras-malhotra you are absolutely correct. either way this is done, there will be something not working "perfectly". but much better to process a job twice than to never process it at all. the risk of doing twice is a little more work for the system on some edge case, but the risk of never doing it is that things will go missing. better to err on the side of caution.

[paras-malhotra]

@moshe-autoleadstar, @lasselehtinen I've submitted a PR for this: #35255

[abellion]

Is there a reason why the locking logic is within Illuminate/Foundation/Bus/PendingDispatch instead of Illuminate\Queue\Queue ? I think it could be moved there in the enqueueUsing method (https://github.com/laravel/framework/blob/8.x/src/Illuminate/Queue/Queue.php#L298).

This would make more sens IMO, and additionally it would allow the feature to work independently from the PendingDispatch class (which is missing from Lumen BTW).

[driesvints]

@abellion if you need this feature then please use Laravel instead of Lumen.

[abellion]

Yeah, sure. But if the locking logic can be moved in the Illuminate\Queue\Queue like I suggested, that would be better. I can make a PR for that. I'm just checking if indeed it can be moved, or if I miss something.

[paras-malhotra]

@abellion the locking needs to happen at the time of dispatch (and not at the time of processing the queue jobs). That's why it's part of the PendingDispatch class.

[abellion]

@paras-malhotra Yep, I think moving it to https://github.com/laravel/framework/blob/8.x/src/Illuminate/Queue/Queue.php#L298 would work :) As I understand this is called just before the job is pushed into the queue.

See https://gist.github.com/abellion/54fa7bda01204d458439cb8a67e2b326

[paras-malhotra]

Yeah @abellion, that gist looks like it should work. Ahh on second thought, it may need to be modified for the dispatch after commit feature to work properly. Perhaps the lock condition needs to be checked before the dispatchAfterCommit condition,

[abellion]

You're right @paras-malhotra, I moved it above !

[oobi]

Hey there -- that edit available in the version of Lumen currently available via Composer or is there some way to update it to grab the new condition? I've just started with Lumen and this solves my current road block!

[oobi]

Is it possible to get a failure notice from that if the job could not be queued?

e.g.
$result = dispatch(new DatabaseDumpJob($db)); // success/fail ?

That sends back a PendingDispatch object which as far as I can tell looks the same whether or not it actually got queued.

[oobi]

Lock needs to use the string 'laravel_unique_job' or it doesn't unlock in CallQueuedHandler.

// > New condition
        if (is_object($job) && $job instanceof ShouldBeUnique) {

            $uniqueId = method_exists($job, 'uniqueId')
                    ? $job->uniqueId()
                    : ($job->uniqueId ?? '');

            $cache = method_exists($job, 'uniqueVia')
                        ? $job->uniqueVia()
                        : $this->container->make(Cache::class);

            $uniqueFor = $job->uniqueFor ?? 0;

            $lock = $cache->lock( 'laravel_unique_job:' . get_class($job) . $uniqueId, $uniqueFor);

            if (true !== $lock->get()) {
                return ;
            }
        }
        // <