Debounced Laravel Jobs

Readme.md

Debounced Laravel Jobs

Runs a specific Job at most once every $delay seconds, and at least once every $maxWait seconds. Uses Laravel's Cache implementation to ensure that this stays true across any number of instances of the application. Works with any existing Job by serializing it and storing it in the Cache.

Example: NOTE: sleep is only for demonstration purposes and should only be found in test code, like the following.

$delay = 15;
$maxWait = 60;

$job = new Jobs\SomeJobToDebounce();

// No matter how many times DebouncedJob is triggered for $job,
// $job will only be handled once (after a 15 second delay).
$this->dispatch(new DebouncedJob($job, $delay, $maxWait));
$this->dispatch(new DebouncedJob($job, $delay, $maxWait));
$this->dispatch(new DebouncedJob($job, $delay, $maxWait));
// etc...

// Wait for the DebouncedJob to trigger so we don't override it below.
sleep($delay+1);

// $job will be handled twice: both after $delay seconds,
// once for the default key prefix and once for 'some_other_pfx'
$this->dispatch(new DebouncedJob($job, $delay, $maxWait));
$this->dispatch(new DebouncedJob($job, $delay, $maxWait));
$this->dispatch(new DebouncedJob($job, $delay, $maxWait, 'some_other_pfx'));
$this->dispatch(new DebouncedJob($job, $delay, $maxWait, 'some_other_pfx'));
// etc...

// Wait for the DebouncedJob to trigger so we don't override it below.
sleep($delay+1);

$countTimes = 100;
// $job will be handled twice.
// Once at 60 seconds ($maxWait),
// and once again at 60 + 55 seconds ($countTimes + $delay)
for ($i = 0; $i < $countTimes; $i++) {
  $this->dispatch(new DebouncedJob($job, $delay, $maxWait));
  sleep(1);
}

sleep($delay+1);

DebouncedJob.php

<?php

namespace App\Support\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;

class DebouncedJob implements ShouldQueue
{
    use \Illuminate\Foundation\Bus\DispatchesJobs;
    use \App\Support\Cache\PrefixedCache;
    use \Illuminate\Queue\SerializesModels;
    use \Illuminate\Queue\InteractsWithQueue;
    use \Illuminate\Bus\Queueable;

    const DEFAULT_PREFIX = 'debounce';
    const KEY_MAX_WAIT = 'maxWait';
    const KEY_DEBOUNCE = 'debounce';

    /**
     * @var array
     */
    protected $cacheKeys = [
        self::KEY_DEBOUNCE,
        self::KEY_MAX_WAIT,
    ];

    /**
     * The Job that is being debounced, in serialized form.
     *
     * @var string
     */
    protected $debounced;

    /**
     * Amount of time (in seconds) to debounce the Job.
     *
     * @var int
     */
    protected $debounce;

    /**
     * Maximum amount of time (in seconds) to wait before this Job gets run.
     *
     * @var int|null
     */
    protected $maxWait;

    /**
     * This can be used to have separately debounced versions of the same Job.
     *
     * @var string
     */
    protected $cachePrefix;

    /**
     * Cache of the unserialized Job instance that was debounced.
     *
     * @var mixed|null
     */
    protected $unserialized;

    public function __construct(
        $debounced,
        $delay,
        $maxWait = null,
        $prefix = self::DEFAULT_PREFIX
    ) {
        $this->debounced = serialize($debounced);
        $this->debounce = $delay;
        $this->maxWait = $maxWait;
        $this->cachePrefix = $prefix;

        $this->cacheDebounceTime();
        $this->cacheMaxWaitTime();
        $this->delay($delay+1);
    }

    /**
     * Handle the Job, if it is time to.
     */
    public function handle()
    {
        // Check if this or a later DebouncedJob instance will handle this task.
        if (!$this->isReadyToHandle()) {
            return;
        }

        // Prevent any future debounced Jobs from being triggered,
        // until a new one is created.
        $this->clearCache();

        // Then, dispatch the original Job that was debounced.
        $this->dispatch(unserialize($this->debounced));
    }

    # Implementation for PrefixedCache
    /**
     * The key to store in the Cache for this Job.
     *
     * @param  string $suffix
     *
     * @return string
     */
    protected function getCacheKey($suffix)
    {
        return sprintf(
            '%s__%s__%s',
            $this->cachePrefix,
            $this->getJobName(),
            $suffix
        );
    }

    private function getJob()
    {
        return $this->unserialized ?:
            $this->unserialized = unserialize($this->debounced);
    }

    private function getJobName()
    {
        $job = $this->getJob();
        if (is_a($job, \Illuminate\Contracts\Queue\Job::class)) {
            return $job->getName();
        }

        return get_class($job);
    }

    /**
     * Determine if the requested Job should be processed immediately.
     *
     * @return boolean
     */
    private function isReadyToHandle()
    {
        $isInPast = function ($_, $cacheKey) {
            return $this->isInPast(
                $this->getCache($cacheKey)
            );
        };

        return (bool) collect($this->cacheKeys)->first($isInPast);
    }

    /**
     * @param  int  $time
     *
     * @return boolean
     */
    private function isInPast($time)
    {
        print "Checking $time...\n";
        return $time && time() >= $time;
    }

    /**
     * Store the debounce time in the Cache.
     */
    private function cacheDebounceTime()
    {
        $this->setCache(static::KEY_DEBOUNCE, time() + $this->debounce);
    }

    /**
     * Store the max wait time in the Cache.
     */
    private function cacheMaxWaitTime()
    {
        if (!$this->maxWait) {
            return;
        }

        if ($this->getCache(static::KEY_MAX_WAIT)) {
            // There is currently a max wait in place,
            // that has not been triggered yet.
            // We don't want to override that value.
            return;
        }

        $this->setCache(static::KEY_MAX_WAIT, time() + $this->maxWait);
    }

    /**
     * Clear all related Cache entries.
     */
    private function clearCache()
    {
        collect($this->cacheKeys)->each(function ($cacheKey) {
            $this->forgetCache($cacheKey);
        });
    }
}

PrefixedCache.php

<?php

namespace App\Support\Cache;

use Cache as CacheFacade;

trait PrefixedCache
{
    /**
     * This prefix is added to all Cache requests.
     *
     * @var string
     */
    protected $cachePrefix = '';

    protected function forgetCache($key)
    {
        return CacheFacade::forget(
            $this->getCacheKey($key)
        );
    }


    protected function getCache($key, $default = null)
    {
        return CacheFacade::get(
            $this->getCacheKey($key),
            $default
        );
    }

    protected function setCache($key, $value)
    {
        return CacheFacade::forever(
            $this->getCacheKey($key),
            $value
        );
    }

    /**
     * Compute the cache key for a given $suffix.
     *
     * @return string
     */
    protected function getCacheKey($suffix)
    {
        if (!$this->cachePrefix) {
            throw new \Exception('No cachePrefix or getCacheKey() defined for PrefixedCache');
        }

        return sprintf(
            '%s__%s',
            $this->cachePrefix,
            $suffix
        );
    }
}

Very nice work! How do you delay the job? I can see you call $this->delay($delay+1); in the code, but the method is not defined (at least not in this class directly). It extends a class which we don't know the definition. Can't find App\Support\Job in Laravel. Could you include it please?

Oops, thanks for pointing that out! My App\Support\Job class was just a convenience class that had the following traits:

    use \Illuminate\Queue\SerializesModels;
    use \Illuminate\Queue\InteractsWithQueue;
    use \Illuminate\Bus\Queueable;

The delay method comes from the Queueable trait. I went ahead and added those traits to the DebouncedJob class here - it should work now.

Cool thanks! Also, another thing I've noticed playing with your code, your are missing only a single use statement in DebouncedJob: use Illuminate\Queue\Jobs\Job; (or use \Illuminate\Contracts\Queue\Job;?) as the Job argument of the constructor is not defined. Everything else seems awesome.

Thanks for this gist. Very useful

Cool thanks! Also, another thing I've noticed playing with your code, your are missing only a single use statement in DebouncedJob: use Illuminate\Queue\Jobs\Job; (or use \Illuminate\Contracts\Queue\Job;?) as the Job argument of the constructor is not defined. Everything else seems awesome.

Thanks for this gist. Very useful

Good catch. That should be a reference to the App\Support\Jobs\Job convenience class that I had, but since I removed that convenience class here, that argument should no longer be typed. I've updated the gist accordingly.

Love this git! But for other readers, this is also helpful https://laravel.com/docs/7.x/cache#managing-locks-across-processes
You can just lock your job down and only when it's done you can release the lock so it can be fired again.

Love this git! But for other readers, this is also helpful https://laravel.com/docs/7.x/cache#managing-locks-across-processes
You can just lock your job down and only when it's done you can release the lock so it can be fired again.

Locks seem like an interesting use case here, but one point of caution about using locks for the particular purpose of this gist: if you try to get a lock on something that's already locked, it will generally halt processing and wait for that lock, which in the best case would be a waist of server resources, and in the worst case, can lead to deadlock. In my opinion, it would be better to use this debounce approach rather than a lock so that you are not giving your queue workers work that is not ready to be dealt with yet. This does depend on your use case though, I can imagine cases where you would definitely want such a lock rather than a debounce. You may even want to combine the two as the lock provides a much stronger guarantee that the same thing won't be processed twice.

Anyone that found this gist looking at debouncing jobs in Laravel queues should probably consider the built-in middleware WithoutOverlapping.

doc: https://laravel.com/docs/9.x/queues#preventing-job-overlaps