New feature: cancel scheduled job

[cyberwolf]

I'm looking at existing PHP libraries for scheduling jobs at a given date or with a given delay. A requirement however is that a job that gets scheduled, can be cancelled later on. phive-queue looks great, but it currently lacks this feature.

Not all storage backends supported by phive-queue support cancelling queued items. To get this implemented, I think a second interface next to Queue will be needed. This interface might be called "CancellableQueue". Implementations that have cancel support can implement the additional methods of this new interface. The unique identifier of a scheduled job could be returned by Queue::push(). Implementations that do not support job IDs can just return void. You could also leave Queue::push() alone, and add a new method getLastPushedId() that returns the ID of the last pushed job.

Are you interested in adopting this feature in phive-queue? My team probably can contribute interfaces and implementations for PDO and Beanstalkd.

[rybakit]

In a project I worked on I had the same requirement to be able to cancel already scheduled jobs. And I solved it in a very simple but effective way which works for any backend. I designed jobs to be responsible for canceling itself based on app-level flags.

For example, let’s say you want to cancel a mail sending for already unsubscribed user. Then the job might look like:

class SendMailJob
{
    private $repository;

    public function __construct(SubscriptionRepository $repository)
    {
        $this->repository = $repository;
    }

    public function __invoke($userId)
    {
         if ($this->repository->isUserUnsubscribed($userId)) {
             return; // cancel job
         }

         // execute job
         $this->sendMail($userId);
    }

    ...    
}

Will this technique work in you case?

[cyberwolf]

Thanks @rybakit for the quick reply.

I've also been thinking about that solution, but at first it looked more complex that way.

Some more background on the wider infrastructure we are operating in, might probably be needed. We're quite heavy into Event Sourcing lately, using the Broadway library as a base for our application. When something isn't provided yet in Broadway we usually look at how it has been implemented in Axon Framework (Java). We thought about using Saga's to control long living process flows, like a process manager. Axon has the concept of an EventScheduler and a ScheduleToken, which we would like to implement on top of phive-queue which seemed like the ideal fit except for the lack of a queued item cancel feature.

One of the use cases we have is scheduled publication. A user can immediately publish an entity, or schedule it for publication at a specific date in the future. After an entity is already scheduled for publication, the item can still be immediately published, or the scheduled date still can be changed.

If phive-queue would provide the cancel feature:

• When the entity gets scheduled for publication, we would queue an item and keep its ID in the Saga state.
• When the entity's scheduled publication is cancelled, we would dequeue the item with the ID stored in the Saga state.
• When the entity's scheduled publication date is changed, we would dequeue the item with the ID stored in the Saga state as well as queue a new item and keep its ID again in the Saga state.
• When an items pops off the queue, the entity gets published.

As you propose, it would also be possible to accomplish this without the cancel feature. At first sight this looked a bit more complicated to me. If we associate the forecasted publication date with a unique ID I think we can accomplish the same end result though:

• When the entity gets scheduled for publication, we generate a unique ID, queue an item containing that ID and put the ID as well in the Saga state.
• When the entity's scheduled publication is cancelled, we would remove the ID from the Saga state.
• When the entity's scheduled publication date is changed, we generate a new unique ID, queue an item containing that ID, and replace the previous ID in the saga state with the new ID.
• When an item pops off the queue, the Saga checks if the item's ID is the same ID stored in its state. If they are the same, the Saga publishes the entity. If they are not the same, nothing happens.

I'll come back to this request if we have other use cases which will be harder to accomplish without the cancellation feature. Thanks already for sharing your thoughts!

[rybakit]

@cyberwolf Thank you too for your interest in this library ;)

I didn’t have a chance to work with saga concept yet, but looking through you explanation I would personally choose the second option, mainly because it looks more robust and has less points of failure to me. If I understand you correctly, every time you enqueue/cancel a job, change a scheduling date or publish an entity, you have to ensure the job id and corresponding saga id are in sync. The point of failure here is that there is a chance that this two-step sync process can fail in the middle, e.g. a job is dequeued, but the saga id is failed to update/remove or vice versa. It will probably also require more code to write and support.

With my approach those issues are not the case, as there is nothing to sync except the very first time when you enqueue a new job, but even then you app will continue to work, as the job will just cancel itself if it doesn't find an associated saga id. But I may be wrong since I don't know too much about sagas and your implementation details.